@@ -0,0 +1,20 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
@pytest.fixture
|
||||
def ascii_schematic():
|
||||
# Purpose-built unit-test records, not an application demo or a real Altium export.
|
||||
return b'''|HEADER=Protel for Windows - Schematic Capture Ascii File Version 5.0|Weight=9
|
||||
|RECORD=31|FontIdCount=1|Size1=10|FontName1=Arial
|
||||
|RECORD=1|LibReference=R_Symbol|Location.X=100|Location.Y=200|CurrentPartId=1|PartCount=2|Orientation=1|IsMirrored=T
|
||||
|RECORD=13|OwnerIndex=1|OwnerPartId=1|Location.X=100|Location.Y=200|Corner.X=100|Corner.Y=220|Color=128
|
||||
|RECORD=2|OwnerIndex=1|OwnerPartId=1|Location.X=100|Location.Y=220|PinLength=10|PinConglomerate=25|Name=A|Designator=1|Electrical=4
|
||||
|RECORD=34|OwnerIndex=1|OwnerPartId=-1|Text=R1|Location.X=105|Location.Y=205
|
||||
|RECORD=41|OwnerIndex=1|OwnerPartId=-1|Name=Comment|Text=4K7|Location.X=110|Location.Y=205
|
||||
|RECORD=41|OwnerIndex=1|OwnerPartId=-1|Name=Manufacturer|Text=Test|IsHidden=T
|
||||
|RECORD=27|OwnerPartId=-1|LocationCount=2|X1=100|Y1=230|X2=150|Y2=230
|
||||
|RECORD=25|Text=VCC|Location.X=150|Location.Y=230
|
||||
'''
|
||||
@@ -0,0 +1,12 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
from app.fonts import fonts,stylesheet
|
||||
|
||||
|
||||
def test_font_manifest_only_allows_explicit_local_paths(tmp_path,monkeypatch):
|
||||
monkeypatch.setenv('FONT_DIR',str(tmp_path)); (tmp_path/'test.woff2').write_bytes(b'font fixture')
|
||||
(tmp_path/'fonts.json').write_text(json.dumps({'Arial':[{'file':'test.woff2','weight':'bold','style':'italic'}]}))
|
||||
assert len(fonts())==1 and "font-weight:bold" in stylesheet()
|
||||
(tmp_path/'fonts.json').write_text(json.dumps({'Arial':'../outside.woff2'}))
|
||||
with pytest.raises(ValueError):fonts()
|
||||
@@ -0,0 +1,43 @@
|
||||
import base64
|
||||
import io
|
||||
import struct
|
||||
import zlib
|
||||
from PIL import Image
|
||||
from app.parser.images import read_images
|
||||
from app.render import svg_document
|
||||
from app.models.schematic import Primitive, Bounds
|
||||
from app.parser.schdoc_parser import parse_schdoc
|
||||
|
||||
|
||||
def storage(payload):
|
||||
header=b'|HEADER=Icon storage\0'
|
||||
return struct.pack('<I',len(header))+header+struct.pack('<I',(1<<24)|len(payload))+payload
|
||||
|
||||
class Ole:
|
||||
def __init__(self,data): self.data=data
|
||||
def __enter__(self): return self
|
||||
def __exit__(self,*args): pass
|
||||
def exists(self,name): return name=='Storage'
|
||||
def openstream(self,name): return io.BytesIO(self.data)
|
||||
|
||||
|
||||
def test_embedded_raster_is_reencoded_and_never_fetches_path(monkeypatch, ascii_schematic):
|
||||
buffer=io.BytesIO(); Image.new('RGB',(2,3),(255,0,0)).save(buffer,format='BMP')
|
||||
compressed=zlib.compress(buffer.getvalue()); filename=b'C:\\Project\\image.bmp'
|
||||
payload=b'\xd0'+bytes([len(filename)])+filename+struct.pack('<I',len(compressed))+compressed
|
||||
monkeypatch.setattr('app.parser.images.olefile.OleFileIO',lambda _:Ole(storage(payload)))
|
||||
assets,errors=read_images(b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1')
|
||||
assert not errors and set(assets)=={'image.bmp'}
|
||||
png=base64.b64decode(assets['image.bmp'].split(',')[1]); assert png.startswith(b'\x89PNG')
|
||||
doc=parse_schdoc(ascii_schematic,'x.SchDoc','x')
|
||||
doc.primitives.append(Primitive(id='image',kind='image',bounds=Bounds(x=0,y=-30,width=20,height=30),object_id='image',image_data=assets['image.bmp']))
|
||||
svg=svg_document(doc)
|
||||
assert b'data:image/png;base64,' in svg and b'C:\\Project' not in svg
|
||||
|
||||
|
||||
def test_bounded_zlib_expansion(monkeypatch):
|
||||
compressed=zlib.compress(b'0'*(16*1024*1024+1)); filename=b'bomb.bmp'
|
||||
payload=b'\xd0'+bytes([len(filename)])+filename+struct.pack('<I',len(compressed))+compressed
|
||||
monkeypatch.setattr('app.parser.images.olefile.OleFileIO',lambda _:Ole(storage(payload)))
|
||||
assets,errors=read_images(b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1')
|
||||
assert not assets and any('limit' in e for e in errors)
|
||||
@@ -0,0 +1,133 @@
|
||||
import math
|
||||
import struct
|
||||
import pytest
|
||||
from app.parser.geometry import point,coordinate,transform_point,color
|
||||
from app.models.schematic import Point
|
||||
from app.parser.record_decoder import detect_format,binary_records,properties,ParseError
|
||||
from app.parser.schdoc_parser import parse_schdoc
|
||||
from app.parser.ole_reader import read_document
|
||||
|
||||
@pytest.mark.parametrize('data,expected',[(b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1','ole'),(b'|HEADER=x','ascii'),(b'\xef\xbb\xbf|HEADER=x','ascii')])
|
||||
def test_detect(data,expected): assert detect_format(data)==expected
|
||||
|
||||
@pytest.mark.parametrize('data',[b'',b'<html/>',b'PK\x03\x04',b'\0'*1024])
|
||||
def test_unknown_format(data):
|
||||
with pytest.raises(ParseError): detect_format(data)
|
||||
|
||||
def test_length_prefix_and_indexing():
|
||||
payloads=[b'|HEADER=Example\0',b'|RECORD=31\0',b'|RECORD=27|X1=42\0']
|
||||
stream=b''.join(struct.pack('<I',len(p))+p for p in payloads)
|
||||
records=binary_records(stream,'FileHeader')
|
||||
assert [r.index for r in records]==[-1,0,1]
|
||||
assert records[2].properties['X1']=='42'
|
||||
|
||||
@pytest.mark.parametrize('data',[b'\x05',struct.pack('<I',100)+b'x',struct.pack('<I',0),struct.pack('<I',0x7fffff)+b'x'])
|
||||
def test_truncated_or_excessive_record(data):
|
||||
with pytest.raises(ParseError): binary_records(data,'FileHeader')
|
||||
|
||||
def test_binary_record_does_not_shift_owners():
|
||||
p=b'|HEADER=x\0';s=struct.pack('<I',len(p))+p+struct.pack('<I',0x1000002)+b'aa'
|
||||
r=binary_records(s,'FileHeader');assert r[1].binary and r[1].index==0
|
||||
|
||||
def test_utf8_precedence_and_case():
|
||||
p=properties('|%UTF8%Comment=مقاومت Ω|Comment=legacy|Text=a=b'.encode())
|
||||
assert p['COMMENT']=='مقاومت Ω' and p['TEXT']=='a=b'
|
||||
|
||||
def test_fraction_and_axis():
|
||||
p={'LOCATION.X':'12','LOCATION.X_FRAC':'50000','LOCATION.Y':'-3','LOCATION.Y_FRAC':'25000'}
|
||||
assert coordinate(p,'LOCATION.X')==12.5
|
||||
assert point(p)==Point(x=12.5,y=2.75)
|
||||
assert color('8388608')=='#000080'
|
||||
|
||||
@pytest.mark.parametrize('rotation,expected',[(0,(12,20)),(90,(10,18)),(180,(8,20)),(270,(10,22))])
|
||||
def test_rotation(rotation,expected):
|
||||
p=transform_point(Point(x=2,y=0),Point(x=10,y=20),rotation)
|
||||
assert p.x==pytest.approx(expected[0]) and p.y==pytest.approx(expected[1])
|
||||
|
||||
def test_mirror_then_rotate():
|
||||
p=transform_point(Point(x=2,y=1),Point(x=10,y=20),90,True)
|
||||
assert p==Point(x=11,y=22)
|
||||
|
||||
def test_component_extraction_and_absolute_geometry(ascii_schematic):
|
||||
d=parse_schdoc(ascii_schematic,'test.SchDoc','test')
|
||||
c=d.components[0]
|
||||
assert (c.designator,c.value,c.comment,c.library_reference)==('R1','4K7','4K7','R_Symbol')
|
||||
assert c.rotation==90 and c.mirrored and c.parameters['Manufacturer']=='Test'
|
||||
assert c.pins[0].number=='1' and c.pins[0].name=='A'
|
||||
assert c.pins[0].connection==Point(x=100,y=-230)
|
||||
# Placed child geometry must not be rotated/mirrored again.
|
||||
line=next(p for p in d.primitives if p.object_id=='FileHeader:2')
|
||||
assert line.path=='M100,-200 L100,-220'
|
||||
assert line.component_id==c.id
|
||||
assert len({x.id for x in d.objects})==len(d.objects)
|
||||
assert set(c.primitive_ids)==set(d.symbols[0].primitive_ids)
|
||||
assert not any(p.text=='Test' for p in d.primitives)
|
||||
|
||||
def test_wire_path_and_connection_vertices(ascii_schematic):
|
||||
d=parse_schdoc(ascii_schematic,'test.SchDoc','test')
|
||||
o=next(o for o in d.objects if o.kind=='wire')
|
||||
p=next(p for p in d.primitives if p.object_id==o.id)
|
||||
assert p.path=='M100,-230 L150,-230'
|
||||
assert o.connection_points==[Point(x=100,y=-230),Point(x=150,y=-230)]
|
||||
|
||||
def test_part_visibility(ascii_schematic):
|
||||
data=ascii_schematic+b'|RECORD=13|OwnerIndex=1|OwnerPartId=2|Location.X=2|Corner.X=3\n'
|
||||
d=parse_schdoc(data,'test.SchDoc','test')
|
||||
assert not d.objects[-1].visible and not d.objects[-1].primitive_ids
|
||||
|
||||
def test_unsupported_preserved(ascii_schematic):
|
||||
d=parse_schdoc(ascii_schematic+b'|RECORD=999|UnknownProperty=important\n','test.SchDoc','test')
|
||||
assert d.objects[-1].properties['UNKNOWNPROPERTY']=='important'
|
||||
assert any(x.code=='unsupported_record' and x.record_type==999 for x in d.diagnostics)
|
||||
|
||||
def test_bad_numeric_geometry_reports_partial(ascii_schematic):
|
||||
d=parse_schdoc(ascii_schematic+b'|RECORD=13|Location.X=nan\n','test.SchDoc','test')
|
||||
assert any(x.code=='malformed_record' and x.severity=='error' for x in d.diagnostics)
|
||||
assert d.objects[-1].properties['LOCATION.X']=='nan'
|
||||
|
||||
def test_unknown_version(ascii_schematic):
|
||||
with pytest.raises(ParseError,match='Unsupported schematic version'):
|
||||
parse_schdoc(ascii_schematic.replace(b'Version 5.0',b'Version 8.0'),'test.SchDoc','test')
|
||||
|
||||
def test_ole_malformed_sector_table():
|
||||
with pytest.raises(ParseError): read_document(b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1'+b'\0'*504)
|
||||
|
||||
def test_owner_cycle(ascii_schematic):
|
||||
d=parse_schdoc(ascii_schematic+b'|RECORD=13|OwnerIndex=9\n','test.SchDoc','test')
|
||||
assert any(x.code=='malformed_record' for x in d.diagnostics)
|
||||
|
||||
@pytest.mark.parametrize('typ,props',[
|
||||
(5,'|LocationCount=4|X1=0|Y1=0|X2=5|Y2=9|X3=10|Y3=9|X4=15|Y4=0'),
|
||||
(7,'|LocationCount=3|X1=0|Y1=0|X2=5|Y2=9|X3=10|Y3=0|IsSolid=T'),
|
||||
(8,'|Radius=5|SecondaryRadius=10'),(9,'|Radius=5|StartAngle=30|EndAngle=180'),
|
||||
(10,'|Corner.X=20|Corner.Y=10|CornerXRadius=2|CornerYRadius=3'),
|
||||
(11,'|Radius=10|SecondaryRadius=5|StartAngle=45|EndAngle=200'),
|
||||
(12,'|Radius=5'),(14,'|Corner.X=10|Corner.Y=20'),
|
||||
(15,'|XSize=100|YSize=60'),(18,'|Name=DATA|Width=50|Style=3'),
|
||||
(22,''),(26,'|LocationCount=2|X1=0|Y1=0|X2=20|Y2=30'),(29,''),
|
||||
(37,'|Corner.X=10|Corner.Y=20'),(225,'|Corner.X=100|Corner.Y=50')])
|
||||
def test_supported_shapes(ascii_schematic,typ,props):
|
||||
d=parse_schdoc(ascii_schematic+f'|RECORD={typ}{props}\n'.encode(),'test.SchDoc','test')
|
||||
obj=d.objects[-1]
|
||||
assert obj.primitive_ids
|
||||
assert not any(x.code=='malformed_record' for x in d.diagnostics)
|
||||
assert all(math.isfinite(p.bounds.width) for p in d.primitives)
|
||||
|
||||
def test_clipped_text_does_not_expand_fit_bounds(ascii_schematic):
|
||||
line='|RECORD=28|Location.X=10|Location.Y=10|Corner.X=60|Corner.Y=40|FontID=1|WordWrap=T|ClipToRect=T|Text='+('Long note text '*100)+'\n'
|
||||
d=parse_schdoc(ascii_schematic+line.encode(),'test.SchDoc','test')
|
||||
obj=d.objects[-1]
|
||||
assert obj.bounds.x>=10 and obj.bounds.x+obj.bounds.width<=60
|
||||
assert obj.bounds.y>=-40 and obj.bounds.y+obj.bounds.height<=-10
|
||||
assert any(p.clip for p in d.primitives if p.object_id==obj.id)
|
||||
|
||||
def test_invalid_record_type_is_preserved(ascii_schematic):
|
||||
d=parse_schdoc(ascii_schematic+b'|RECORD=abc|OriginalField=value\n','test.SchDoc','test')
|
||||
assert d.objects[-1].properties['ORIGINALFIELD']=='value'
|
||||
assert any(w.code=='malformed_record' for w in d.diagnostics)
|
||||
|
||||
def test_implementation_parameter_is_not_drawn(ascii_schematic):
|
||||
data=ascii_schematic+b'|RECORD=48|OwnerIndex=1\n|RECORD=41|OwnerIndex=9|Name=ModelParam|Text=must-not-render\n'
|
||||
d=parse_schdoc(data,'test.SchDoc','test')
|
||||
assert not d.objects[-1].visible
|
||||
assert not any(p.text=='must-not-render' for p in d.primitives)
|
||||
@@ -0,0 +1,165 @@
|
||||
import json
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import select, update
|
||||
from app.config import settings
|
||||
from app.db import Job, claim, engine, sessions
|
||||
from app.main import app
|
||||
from app.storage import storage
|
||||
from app.worker import process
|
||||
|
||||
KEY = 'test-api-key-' + 'x'*40
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv('STORAGE_DIR', str(tmp_path))
|
||||
monkeypatch.setenv('DATABASE_URL', f'sqlite:///{tmp_path}/test.db')
|
||||
monkeypatch.setenv('API_KEY', KEY)
|
||||
monkeypatch.setenv('PUBLIC_BASE_URL', 'http://testserver')
|
||||
monkeypatch.setenv('MAX_UPLOAD_BYTES', '1048576')
|
||||
settings.cache_clear(); storage.cache_clear(); sessions.cache_clear(); engine.cache_clear()
|
||||
with TestClient(app) as client: yield client
|
||||
engine().dispose()
|
||||
settings.cache_clear(); storage.cache_clear(); sessions.cache_clear(); engine.cache_clear()
|
||||
|
||||
|
||||
def upload(client, data, **fields):
|
||||
return client.post('/api/v1/schematics/render', files={'file': ('sample.SchDoc', data, 'application/octet-stream')},
|
||||
data=fields, headers={'X-API-Key': KEY})
|
||||
|
||||
|
||||
def test_private_upload_worker_viewer_and_expiry(client, ascii_schematic):
|
||||
response = upload(client, ascii_schematic)
|
||||
assert response.status_code == 202, response.text
|
||||
body = response.json(); id = body['id']; token = {'X-Viewer-Token': body['viewer_token']}
|
||||
assert len(id) == 48 and '#token=' in body['viewer_url']
|
||||
base = f'/api/v1/schematics/{id}'
|
||||
assert client.get(base+'/status').status_code == 404
|
||||
assert client.get(base+'/status',headers=token).json()['status'] == 'queued'
|
||||
assert client.get(base+'/data',headers=token).status_code == 409
|
||||
job = claim(); assert job['id'] == id
|
||||
process(job)
|
||||
response = client.get(base+'/status',headers=token)
|
||||
assert response.json()['status'] == 'completed', response.text
|
||||
assert response.json()['compatibility'] == 'partial'
|
||||
data = client.get(base+'/data',headers=token).json()
|
||||
assert data['components'][0]['designator'] == 'R1'
|
||||
drawing = client.get(base+'/svg',headers=token)
|
||||
assert drawing.status_code == 200 and '<svg' in drawing.text and 'R1' in drawing.text
|
||||
view = client.get('/view/'+id)
|
||||
assert view.status_code == 200 and 'viewer.js' in view.text
|
||||
assert 'no-referrer' == view.headers['referrer-policy']
|
||||
with sessions().begin() as db: db.execute(update(Job).where(Job.id==id).values(expires_at=time.time()-1))
|
||||
assert client.get(base+'/data',headers=token).status_code == 410
|
||||
assert client.get(base+'/status').status_code == 404
|
||||
|
||||
|
||||
def test_unlisted_and_api_key(client, ascii_schematic):
|
||||
assert client.post('/api/v1/schematics/render').status_code == 401
|
||||
assert client.post('/api/v1/schematics/render',headers={'X-API-Key':'bad'}).status_code == 401
|
||||
body = upload(client,ascii_schematic,access='unlisted').json()
|
||||
assert 'viewer_token' not in body
|
||||
assert client.get(f"/api/v1/schematics/{body['id']}/status").status_code == 200
|
||||
|
||||
|
||||
def test_upload_validation(client):
|
||||
assert upload(client,b'not a schematic').status_code == 422
|
||||
response=client.post('/api/v1/schematics/render',files={'file':('bad.PcbDoc',b'a')},headers={'X-API-Key':KEY})
|
||||
assert response.status_code == 415
|
||||
assert upload(client,b'x'*(1048576+1)).status_code == 413
|
||||
assert upload(client,b'|HEADER=x',expires_in=999999999).status_code == 422
|
||||
assert upload(client,b'|HEADER=x',access='public').status_code == 422
|
||||
|
||||
|
||||
def test_multipart_limit_before_spooling(client):
|
||||
r = client.post('/api/v1/schematics/render',content=b'x'*(3*1048576),
|
||||
headers={'X-API-Key':KEY,'Content-Type':'multipart/form-data; boundary=foo'})
|
||||
assert r.status_code == 413, r.text
|
||||
|
||||
|
||||
def test_corruption_is_durable_failure(client):
|
||||
body = upload(client,b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1'+b'\0'*504).json()
|
||||
process(claim())
|
||||
r = client.get(f"/api/v1/schematics/{body['id']}/status",headers={'X-API-Key':KEY})
|
||||
assert r.json()['status'] == 'failed'
|
||||
assert r.json()['error']['code'] == 'invalid_schematic'
|
||||
|
||||
|
||||
def test_queue_capacity(client, ascii_schematic, monkeypatch):
|
||||
monkeypatch.setenv('MAX_PENDING_JOBS','1'); settings.cache_clear()
|
||||
assert upload(client,ascii_schematic).status_code == 202
|
||||
assert upload(client,ascii_schematic).status_code == 503
|
||||
|
||||
|
||||
def test_atomic_claim_and_stale_worker_fencing(client, ascii_schematic):
|
||||
body=upload(client,ascii_schematic).json()
|
||||
with ThreadPoolExecutor(max_workers=4) as pool: jobs=list(pool.map(lambda _:claim(),range(4)))
|
||||
found=[job for job in jobs if job]
|
||||
assert len(found)==1
|
||||
old=found[0]
|
||||
with sessions().begin() as db: db.execute(update(Job).where(Job.id==body['id']).values(lease_until=time.time()-1))
|
||||
new=claim(); assert new['attempt']==2
|
||||
process(new); process(old)
|
||||
with sessions()() as db:
|
||||
job=db.get(Job,body['id'])
|
||||
assert job.status=='completed' and job.output_prefix.endswith('attempt-2')
|
||||
|
||||
|
||||
def test_storage_path_traversal(client):
|
||||
for key in ['../etc/passwd','a'*48+'/../input.SchDoc','/tmp/evil','a'*48+'/input.SchDoc/evil']:
|
||||
with pytest.raises(ValueError): storage().get(key)
|
||||
|
||||
|
||||
def test_api_docs_and_static(client):
|
||||
assert client.get('/openapi.json').json()['info']['version']=='1.0.0'
|
||||
assert client.get('/').status_code==200
|
||||
assert client.get('/assets/viewer.js').status_code==200
|
||||
assert client.get('/assets/not-allowed.js').status_code==404
|
||||
assert client.get('/view/nope').status_code==404
|
||||
|
||||
|
||||
def test_production_fails_closed(client, monkeypatch):
|
||||
monkeypatch.setenv('APP_ENV','production'); monkeypatch.setenv('API_KEY',''); settings.cache_clear()
|
||||
with pytest.raises(RuntimeError,match='API_KEY'): settings()
|
||||
monkeypatch.setenv('API_KEY',KEY); settings.cache_clear()
|
||||
with pytest.raises(RuntimeError,match='HTTPS'): settings()
|
||||
|
||||
|
||||
def test_expired_storage_cleanup(client, ascii_schematic):
|
||||
from app.cleanup import cleanup
|
||||
body=upload(client,ascii_schematic).json(); process(claim())
|
||||
identifier=body['id']
|
||||
with sessions().begin() as db: db.execute(update(Job).where(Job.id==identifier).values(expires_at=time.time()-1))
|
||||
assert cleanup()==1
|
||||
with sessions()() as db: assert db.get(Job,identifier) is None
|
||||
with pytest.raises(FileNotFoundError): storage().get(identifier+'/input.SchDoc')
|
||||
|
||||
|
||||
def test_expired_worker_is_not_reclaimed(client, ascii_schematic):
|
||||
body=upload(client,ascii_schematic).json()
|
||||
with sessions().begin() as db: db.execute(update(Job).where(Job.id==body['id']).values(expires_at=time.time()-1))
|
||||
assert claim() is None
|
||||
|
||||
|
||||
def test_repeated_worker_loss_fails_job(client, ascii_schematic):
|
||||
body=upload(client,ascii_schematic).json(); assert claim()['attempt']==1
|
||||
for attempt in (2,3):
|
||||
with sessions().begin() as db: db.execute(update(Job).where(Job.id==body['id']).values(lease_until=time.time()-1))
|
||||
result=claim()
|
||||
if attempt==2: assert result['attempt']==2
|
||||
else: assert result is None
|
||||
with sessions()() as db: assert db.get(Job,body['id']).error_code=='worker_lost'
|
||||
|
||||
|
||||
def test_parser_wall_timeout_is_reported(client,ascii_schematic,monkeypatch):
|
||||
import subprocess
|
||||
from app import worker
|
||||
body=upload(client,ascii_schematic).json(); job=claim()
|
||||
def timeout(*args,**kw): raise subprocess.TimeoutExpired('parser',60)
|
||||
monkeypatch.setattr(worker.subprocess,'run',timeout)
|
||||
process(job)
|
||||
status=client.get(f"/api/v1/schematics/{body['id']}/status",headers={'X-API-Key':KEY}).json()
|
||||
assert status['status']=='failed' and status['error']['code']=='parse_timeout'
|
||||
@@ -0,0 +1,67 @@
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
from xml.etree.ElementTree import fromstring
|
||||
import pytest
|
||||
from PIL import Image
|
||||
from app.parser.schdoc_parser import parse_schdoc
|
||||
from app.render import svg_document
|
||||
|
||||
|
||||
def test_svg_escapes_untrusted_content(ascii_schematic):
|
||||
malicious=b'|RECORD=4|Location.X=1|Text=<script>alert("x")</script>&\n'
|
||||
doc=parse_schdoc(ascii_schematic+malicious,'x.SchDoc','x')
|
||||
svg=svg_document(doc); tree=fromstring(svg)
|
||||
assert not list(tree.iter('{http://www.w3.org/2000/svg}script'))
|
||||
assert b'<script>' in svg
|
||||
assert b'viewBox="0.0 -760.0 1150.0 760.0"' in svg
|
||||
|
||||
|
||||
def test_custom_page_and_rotation(ascii_schematic):
|
||||
raw=ascii_schematic.replace(b'FontIdCount=1',b'UseCustomSheet=T|CustomX=500|CustomY=900|FontIdCount=1')
|
||||
doc=parse_schdoc(raw,'x.SchDoc','x')
|
||||
assert doc.sheet.bounds.width==500 and doc.sheet.bounds.height==900
|
||||
assert b'rotate(-90.0)' in svg_document(doc)
|
||||
|
||||
|
||||
def test_no_substituted_unsupported_power_symbol(ascii_schematic):
|
||||
raw=ascii_schematic+b'|RECORD=17|Style=6|Location.X=55|Text=EARTH|ShowNetName=T\n'
|
||||
doc=parse_schdoc(raw,'x.SchDoc','x')
|
||||
assert any(d.code=='power_style' for d in doc.diagnostics)
|
||||
assert all(p.kind=='text' for p in doc.primitives if p.object_id==doc.objects[-1].id)
|
||||
|
||||
|
||||
FIXTURE=Path(__file__).resolve().parents[2]/'.runtime/fixtures/Channel.SchDoc'
|
||||
|
||||
@pytest.mark.parametrize('path',[FIXTURE, Path(os.getenv('SCHDOC_SAMPLE', '/nonexistent'))])
|
||||
def test_real_binary_fixture(path):
|
||||
if not path.is_file(): pytest.skip('Fetch the native fixture or set SCHDOC_SAMPLE')
|
||||
doc=parse_schdoc(path.read_bytes(),path.name,'real')
|
||||
assert doc.metadata['format']=='ole'
|
||||
assert len(doc.components)>0 and len(doc.primitives)>5
|
||||
assert all(o.properties for o in doc.objects)
|
||||
import cairosvg
|
||||
png=cairosvg.svg2png(bytestring=svg_document(doc),output_width=1600)
|
||||
with Image.open(io.BytesIO(png)) as image:
|
||||
assert image.width==1600 and image.height>0
|
||||
|
||||
|
||||
def test_mirrored_parameter_placement_keeps_glyphs_readable(ascii_schematic):
|
||||
raw=ascii_schematic+b'|RECORD=41|Text=DDR3[7A]|IsMirrored=T|Justification=5|Location.X=1077|Location.Y=781\n'
|
||||
doc=parse_schdoc(raw,'x.SchDoc','x')
|
||||
text=next(p for p in doc.primitives if p.text=='DDR3[7A]')
|
||||
assert not text.mirrored and text.anchor=='end' and text.position.x==1077
|
||||
|
||||
|
||||
def test_svg_keeps_fractional_native_coordinates(ascii_schematic):
|
||||
raw=ascii_schematic+b'|RECORD=13|Location.X=1000|Location.X_FRAC=12345|Location.Y=500|Location.Y_FRAC=67890|Corner.X=1001|Corner.Y=500\n'
|
||||
doc=parse_schdoc(raw,'x.SchDoc','x')
|
||||
primitive=next(p for p in doc.primitives if p.object_id==doc.objects[-1].id)
|
||||
assert primitive.path=='M1000.12345,-500.6789 L1001,-500'
|
||||
|
||||
|
||||
def test_rounded_rectangle_preserves_elliptical_corner_radii(ascii_schematic):
|
||||
raw=ascii_schematic+b'|RECORD=10|Corner.X=20|Corner.Y=10|CornerXRadius=2|CornerYRadius=3\n'
|
||||
doc=parse_schdoc(raw,'x.SchDoc','x')
|
||||
primitive=next(p for p in doc.primitives if p.object_id==doc.objects[-1].id)
|
||||
assert primitive.path.count('A2.0,3.0')==4 and ' Q' not in primitive.path
|
||||
@@ -0,0 +1,23 @@
|
||||
"""Protocol-level S3 contract check with botocore's validation stub (no live S3)."""
|
||||
import io
|
||||
import boto3
|
||||
from botocore.response import StreamingBody
|
||||
from botocore.stub import Stubber
|
||||
from app.storage import Storage
|
||||
|
||||
|
||||
def test_s3_private_round_trip_contract(tmp_path,monkeypatch):
|
||||
from app.config import settings
|
||||
from app.storage import storage
|
||||
monkeypatch.setenv('STORAGE_DIR',str(tmp_path)); monkeypatch.setenv('S3_BUCKET','test-private')
|
||||
settings.cache_clear(); storage.cache_clear()
|
||||
client=boto3.client('s3',region_name='us-east-1',aws_access_key_id='test',aws_secret_access_key='test')
|
||||
monkeypatch.setattr(boto3,'client',lambda *a,**kw:client)
|
||||
key='a'*48+'/input.SchDoc'; payload=b'actual bytes'
|
||||
with Stubber(client) as stub:
|
||||
stub.add_response('put_object',{}, {'Bucket':'test-private','Key':key,'Body':payload,'ContentType':'application/octet-stream'})
|
||||
stub.add_response('get_object',{'Body':StreamingBody(io.BytesIO(payload),len(payload))},{'Bucket':'test-private','Key':key})
|
||||
stub.add_response('delete_object',{}, {'Bucket':'test-private','Key':key})
|
||||
service=Storage();service.put(key,payload);assert service.get(key)==payload;service.delete(key)
|
||||
stub.assert_no_pending_responses()
|
||||
settings.cache_clear(); storage.cache_clear()
|
||||
@@ -0,0 +1,29 @@
|
||||
"""Validate the regression harness; these synthetic images are not Altium references."""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
ROOT=Path(__file__).resolve().parents[2]
|
||||
|
||||
def test_visual_harness_detects_pixel_changes(tmp_path,ascii_schematic):
|
||||
source=tmp_path/'unit.SchDoc';source.write_bytes(ascii_schematic)
|
||||
first=tmp_path/'initial'
|
||||
command=[sys.executable,str(ROOT/'scripts/validate.py'),str(source)]
|
||||
run=subprocess.run(command+['--width','400','--output',str(first)],capture_output=True,text=True)
|
||||
assert run.returncode==0,run.stderr
|
||||
reference=first/'rendered.png'
|
||||
equal=tmp_path/'equal'
|
||||
run=subprocess.run(command+['--reference',str(reference),'--output',str(equal),'--max-mae','0','--max-changed-fraction','0'],capture_output=True,text=True)
|
||||
assert run.returncode==0,run.stderr
|
||||
report=json.loads((equal/'report.json').read_text())
|
||||
assert report['mae']==0 and report['regression_passed'] and not report['fidelity_verified']
|
||||
image=Image.open(reference).convert('RGB');ImageDraw.Draw(image).rectangle((0,0,80,80),fill='black')
|
||||
altered=tmp_path/'changed.png';image.save(altered)
|
||||
changed=tmp_path/'different'
|
||||
run=subprocess.run(command+['--reference',str(altered),'--output',str(changed)],capture_output=True,text=True)
|
||||
assert run.returncode==1,run.stderr
|
||||
assert not json.loads((changed/'report.json').read_text())['regression_passed']
|
||||
assert (changed/'difference.png').exists() and (changed/'overlay.png').exists()
|
||||
مرجع در شماره جدید
مسدود کردن کاربر