299 خطوط
16 KiB
Python
299 خطوط
16 KiB
Python
import copy
|
|
import json
|
|
import os
|
|
import struct
|
|
import time
|
|
from pathlib import Path
|
|
from xml.etree.ElementTree import fromstring
|
|
import pytest
|
|
from sqlalchemy import update
|
|
from app.documents import parse_document
|
|
from app.parser.record_decoder import ParseError
|
|
from app.parser.pcb_binary import Reader, at, binary_records, wide_strings
|
|
from app.parser.pcb_container import identify_pcb
|
|
from app.parser.pcb_geometry import unit,raw_unit,pad_shape
|
|
from app.models.schematic import Point
|
|
from app.render import svg_document
|
|
from app.pcb_diff import compare_pcbs
|
|
from app.db import Job,claim,sessions
|
|
from app.storage import storage
|
|
from app.worker import process
|
|
from pcb_fixture import pcb
|
|
from test_platform import client,KEY,upload
|
|
|
|
|
|
def parse(raw):return parse_document(raw,'board.PcbDoc','a'*48,document_type='pcb')
|
|
def create(client,raw,compare=False,**fields):
|
|
files={'old_file':('old.PcbDoc',raw),'new_file':('new.PcbDoc',fields.pop('new',raw))} if compare else {'file':('board.PcbDoc',raw)}
|
|
return client.post('/api/v1/pcbs/comparisons' if compare else '/api/v1/pcbs/render',headers={'X-API-Key':KEY},files=files,data=fields)
|
|
|
|
|
|
def test_pcb_units_and_checked_framing():
|
|
assert raw_unit(100000)==1 and unit('2.54mm')==pytest.approx(10)
|
|
for bad in ['nan','infinity','999999999mm']:
|
|
with pytest.raises(ValueError):unit(bad)
|
|
with pytest.raises(ParseError):Reader(b'\x01').read('I')
|
|
with pytest.raises(ParseError):at(struct.pack('<d',float('nan')),0,'d')
|
|
with pytest.raises(ParseError):list(binary_records(b'\x04\xff\xff\xff\x7f','Tracks6',4))
|
|
value='مقاومت Ω\0'.encode('utf-16le')
|
|
assert wide_strings(struct.pack('<II',1,len(value))+value)[1]=='مقاومت Ω'
|
|
for shape in (1,2,3,9):
|
|
path,box=pad_shape(Point(x=10,y=-20),8,4,shape,90,30)
|
|
assert path.endswith('Z') and box.width==pytest.approx(4) and box.height==pytest.approx(8)
|
|
|
|
|
|
def test_native_fixture_structure_geometry_and_layers():
|
|
raw=pcb();identify_pcb(raw);doc=parse(raw)
|
|
track=next(o for o in doc.objects if o.kind=='track')
|
|
assert track.connection_points==[Point(x=1,y=-1),Point(x=3,y=-1)]
|
|
assert track.net_id==doc.nets[0].id and track.properties['NET_NAME']=='GND'
|
|
assert doc.metadata['copper_stack']==['1','32']
|
|
tree=fromstring(svg_document(doc))
|
|
assert tree.attrib['aria-label']=='PCB'
|
|
assert any(node.attrib.get('data-layer-id')=='1' for node in tree.iter())
|
|
|
|
|
|
@pytest.mark.parametrize('raw',[b'garbage',b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1'+b'\0'*504,pcb(version='9.0')])
|
|
def test_invalid_envelopes(raw):
|
|
with pytest.raises(ParseError):identify_pcb(raw)
|
|
|
|
|
|
def test_structural_diff_movement_width_and_no_mutation():
|
|
old=parse(pcb());same=parse(pcb());before=old.model_dump_json()
|
|
assert compare_pcbs(old,same)['total_changed']==0
|
|
moved=compare_pcbs(old,parse(pcb(x=200000)))
|
|
assert moved['summary']==dict(added=0,removed=0,modified=0,moved=1)
|
|
assert moved['changes'][0]['movement']==dict(dx=1,dy=0)
|
|
modified=compare_pcbs(old,parse(pcb(width=200000)))
|
|
assert modified['summary']['modified']==1
|
|
assert compare_pcbs(old,parse(pcb(track=False)))['summary']['removed']==1
|
|
assert compare_pcbs(old,parse(pcb(net='VCC')))['summary']['modified']==1
|
|
assert old.model_dump_json()==before
|
|
|
|
|
|
@pytest.mark.parametrize('comparison',[False,True])
|
|
def test_pcb_api_worker_capabilities_expiry_cleanup(client,comparison):
|
|
response=create(client,pcb(),comparison,new=pcb(width=200000)) if comparison else create(client,pcb())
|
|
assert response.status_code==202,response.text
|
|
result=response.json();identifier=result['id'];headers={'X-Viewer-Token':result['viewer_token']}
|
|
base=f'/api/v1/pcbs/comparisons/{identifier}' if comparison else f'/api/v1/pcbs/{identifier}'
|
|
status=base if comparison else base+'/status'
|
|
svg=base+'/svg/old' if comparison else base+'/svg'
|
|
assert 'summary' not in result and '#token=' in result['viewer_url']
|
|
assert client.get(status).status_code==404
|
|
assert client.get(status,headers={'X-Viewer-Token':'wrong'}).status_code==404
|
|
assert client.get(svg,headers=headers).status_code==409
|
|
process(claim());result=client.get(status,headers=headers).json()
|
|
assert result['status']=='completed',result
|
|
assert client.get(svg,headers=headers).status_code==200
|
|
if comparison:assert result['summary']['modified']==1
|
|
else:
|
|
data=client.get(base+'/data',headers=headers).json()
|
|
assert data['document_type']=='pcb'
|
|
for path in (f'/view/pcb/{identifier}',f'/compare/pcb/{identifier}'):
|
|
response=client.get(path);assert response.status_code==200
|
|
assert len(response.headers.get_list('content-security-policy'))==1
|
|
assert "frame-ancestors 'self' https://git.sadentel.net" in response.headers['content-security-policy']
|
|
assert 'x-frame-options' not in response.headers
|
|
with sessions().begin() as db:db.execute(update(Job).where(Job.id==identifier).values(expires_at=time.time()-1))
|
|
assert client.get(status,headers=headers).status_code==410
|
|
from app.cleanup import cleanup
|
|
assert cleanup()==1
|
|
with pytest.raises(FileNotFoundError):storage().get(identifier+('/pcb-comparison.json' if comparison else '/input.PcbDoc'))
|
|
if comparison:
|
|
with pytest.raises(FileNotFoundError):storage().get(identifier+'/old.PcbDoc')
|
|
|
|
|
|
@pytest.mark.parametrize('access',['private','unlisted'])
|
|
def test_all_four_job_kinds_are_isolated_even_for_service_key(client,ascii_schematic,access):
|
|
from test_comparison import create as schematic_compare
|
|
schematic=upload(client,ascii_schematic,access=access).json()
|
|
sch_compare=schematic_compare(client,ascii_schematic,ascii_schematic,access=access).json()
|
|
board=create(client,pcb(),access=access).json();comparison=create(client,pcb(),True,access=access).json()
|
|
routes=[lambda id:f'/api/v1/schematics/{id}/status',lambda id:f'/api/v1/comparisons/{id}',
|
|
lambda id:f'/api/v1/pcbs/{id}/status',lambda id:f'/api/v1/pcbs/comparisons/{id}']
|
|
for i,item in enumerate([schematic,sch_compare,board,comparison]):
|
|
for j,route in enumerate(routes):
|
|
for headers in ({'X-API-Key':KEY},{'X-Viewer-Token':item.get('viewer_token','')},{}):
|
|
response=client.get(route(item['id']),headers=headers)
|
|
if i!=j:assert response.status_code==404,(i,j,response.text)
|
|
elif 'X-API-Key' in headers or access=='unlisted' or headers.get('X-Viewer-Token'):assert response.status_code==200
|
|
|
|
|
|
def test_auth_cors_upload_validation_and_body_limits(client,ascii_schematic):
|
|
for path in ('/api/v1/pcbs/render','/api/v1/pcbs/comparisons'):
|
|
assert client.post(path,content=b'invalid multipart').status_code==401
|
|
preflight=client.options(path,headers={'Origin':'https://git.sadentel.net','Access-Control-Request-Method':'POST','Access-Control-Request-Headers':'x-api-key'})
|
|
assert preflight.status_code==200 and preflight.headers['access-control-allow-origin']=='https://git.sadentel.net'
|
|
assert client.post(path,content=b'x'*(4*1048576),headers={'X-API-Key':KEY,'Content-Type':'multipart/form-data; boundary=x'}).status_code==413
|
|
assert create(client,ascii_schematic).status_code==422
|
|
assert upload(client,pcb()).status_code==422
|
|
assert create(client,pcb(version='9.0')).status_code==422
|
|
assert create(client,pcb(),expires_in=999999999).status_code==422
|
|
assert client.post('/api/v1/pcbs/render',headers={'X-API-Key':KEY},files={'file':('wrong.SchDoc',pcb())}).status_code==415
|
|
|
|
|
|
def test_failed_decode_and_second_storage_write_cleanup(client,monkeypatch):
|
|
created=create(client,pcb(corrupt=True)).json();process(claim())
|
|
result=client.get(created['status_url'],headers={'X-Viewer-Token':created['viewer_token']}).json()
|
|
assert result['status']=='failed' and result['error']['code']=='invalid_pcb'
|
|
original=storage().put;keys=[]
|
|
def fail(key,body,*args):
|
|
keys.append(key);original(key,body,*args)
|
|
if len(keys)==2:raise OSError('write acknowledged late')
|
|
monkeypatch.setattr(storage(),'put',fail)
|
|
assert create(client,pcb(),True).status_code==503
|
|
assert all(not storage().path(key).exists() for key in keys)
|
|
assert claim() is None
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def real_board():
|
|
sample=os.getenv('PCBDOC_SAMPLE')
|
|
if not sample:pytest.skip('Set PCBDOC_SAMPLE to a private real PCB fixture')
|
|
path=Path(sample)
|
|
return parse_document(path.read_bytes(),path.name,'a'*48,document_type='pcb')
|
|
|
|
|
|
def test_real_board_and_self_comparison(real_board):
|
|
doc=real_board
|
|
assert len(doc.components)>0 and len(doc.primitives)>0
|
|
assert len({p.id for p in doc.primitives})==len(doc.primitives)
|
|
assert compare_pcbs(doc,doc)['total_changed']==0
|
|
assert any(p.role=='pad' for p in doc.primitives)
|
|
assert any(p.role=='drill' for p in doc.primitives)
|
|
assert all(set(o.layer_ids)<={layer.id for layer in doc.layers} for o in doc.objects)
|
|
|
|
|
|
def test_pad_stack_slot_via_layers_and_native_ids():
|
|
from pcb_fixture import rich_pcb
|
|
doc=parse(rich_pcb());pad=next(o for o in doc.objects if o.kind=='pad');via=next(o for o in doc.objects if o.kind=='via')
|
|
assert pad.native_uid=='PAD_UID' and pad.component_id==doc.components[0].id
|
|
assert pad.layer_ids==['1','32'] and via.layer_ids==['1','32']
|
|
pads=[p for p in doc.primitives if p.object_id==pad.id and p.role=='pad']
|
|
assert {p.layer_id for p in pads}=={'1','32'}
|
|
assert next(p for p in pads if p.layer_id=='1').bounds.width==1
|
|
assert next(p for p in pads if p.layer_id=='32').bounds.width==2
|
|
slot=next(p for p in doc.primitives if p.object_id==pad.id and p.role=='drill')
|
|
assert slot.bounds.width==1 and slot.bounds.height==.5 and slot.layer_ids==['1','32']
|
|
bottom_via=next(p for p in doc.primitives if p.object_id==via.id and p.layer_id=='32')
|
|
assert bottom_via.bounds.width==2
|
|
|
|
|
|
def test_footprint_move_rotation_and_child_edits_are_one_change():
|
|
from pcb_fixture import rich_pcb
|
|
from app.parser.pcb_geometry import pad_shape
|
|
old=parse(rich_pcb());new=old.model_copy(deep=True)
|
|
component=new.components[0];component.position.x+=4
|
|
for o in new.objects:
|
|
if o.component_id==component.id:o.position.x+=4;o.bounds.x+=4
|
|
# Paths translated independently from their existing SVG string.
|
|
for p in new.primitives:
|
|
if p.component_id==component.id:
|
|
if p.role=='pad':
|
|
w,h=(1,2) if p.layer_id=='1' else (2,3)
|
|
p.path,p.bounds=pad_shape(Point(x=5,y=-1),w,h,1 if p.layer_id=='1' else 3,0)
|
|
elif p.role=='drill':p.path,p.bounds=pad_shape(Point(x=5,y=-1),1,.5,1,0)
|
|
# Source parameters retain physical placement too.
|
|
component.parameters['X']='50mil'
|
|
result=compare_pcbs(old,new)
|
|
assert result['summary']==dict(added=0,removed=0,modified=0,moved=1),result['changes']
|
|
component.value='new value'
|
|
result=compare_pcbs(old,new)
|
|
assert result['summary']['modified']==1 and result['changes'][0]['moved']
|
|
# Reparse native revision with a rotated footprint/pad: only one footprint modification.
|
|
rotated=parse(rich_pcb());rotated.components[0].rotation=90
|
|
assert compare_pcbs(old,rotated)['summary']['modified']==1
|
|
|
|
|
|
def test_record_reordering_net_renumber_and_duplicate_ambiguity():
|
|
import io
|
|
import olefile
|
|
from pcb_fixture import cfb,text,record
|
|
def native(reverse):
|
|
with olefile.OleFileIO(io.BytesIO(pcb())) as ole:streams={'/'.join(e):ole.openstream(e).read() for e in ole.listdir()}
|
|
streams['Nets6/Data']=text({'NAME':'VCC'})+text({'NAME':'GND'}) if reverse else text({'NAME':'GND'})+text({'NAME':'VCC'})
|
|
records=[]
|
|
for x,net in [(100000,0),(900000,1)]:
|
|
p=bytearray(45);p[0]=1
|
|
struct.pack_into('<HHH',p,3,(1-net) if reverse else net,65535,65535)
|
|
struct.pack_into('<iiiii',p,13,x,100000,x+200000,100000,100000)
|
|
records.append(record(4,p))
|
|
streams['Tracks6/Data']=b''.join(reversed(records) if reverse else records)
|
|
return parse(cfb(streams))
|
|
assert compare_pcbs(native(False),native(True))['total_changed']==0
|
|
old=native(False);new=old.model_copy(deep=True)
|
|
track=next(o for o in new.objects if o.kind=='track').model_copy(deep=True);track.id+='-duplicate';new.objects.append(track)
|
|
assert compare_pcbs(old,new)['summary']['added']==1
|
|
|
|
|
|
@pytest.mark.parametrize('path',['/view/pcb/invalid','/compare/pcb/invalid'])
|
|
def test_pcb_error_page_csp(client,path):
|
|
r=client.get(path);assert r.status_code==404
|
|
assert len(r.headers.get_list('content-security-policy'))==1
|
|
assert "frame-ancestors 'self' https://git.sadentel.net" in r.headers['content-security-policy']
|
|
assert 'x-frame-options' not in r.headers
|
|
|
|
|
|
def test_pcb_admission_limit_and_stale_attempt_fencing(client,monkeypatch):
|
|
from app.config import settings
|
|
monkeypatch.setenv('MAX_PENDING_JOBS','1');settings.cache_clear()
|
|
result=create(client,pcb(),access='unlisted').json()
|
|
assert create(client,pcb()).status_code==503
|
|
old=claim()
|
|
with sessions().begin() as db:db.execute(update(Job).where(Job.id==result['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,result['id']);assert job.status=='completed' and job.output_prefix.endswith('/attempt-2')
|
|
|
|
|
|
def test_cutout_is_not_dropped_as_an_unowned_region():
|
|
import io
|
|
import olefile
|
|
from pcb_fixture import cfb,record
|
|
with olefile.OleFileIO(io.BytesIO(pcb())) as ole:streams={'/'.join(e):ole.openstream(e).read() for e in ole.listdir()}
|
|
p=bytearray(18);p[0]=1;struct.pack_into('<HHH',p,3,65535,65535,65535)
|
|
props=b'|KIND=0|ISBOARDCUTOUT=TRUE\0'
|
|
p+=struct.pack('<I',len(props))+props+struct.pack('<I',3)
|
|
for x,y in [(200000,200000),(400000,200000),(400000,400000),(200000,400000)]:
|
|
p+=struct.pack('<Biiiiidd',0,x,y,0,0,0,0.,0.)
|
|
streams['ShapeBasedRegions6/Data']=record(11,p)
|
|
doc=parse(cfb(streams));cutout=next(p for p in doc.primitives if p.role=='board_cutout')
|
|
assert cutout.layer_id=='0' and cutout.bounds.width==2 and cutout.bounds.height==2
|
|
assert cutout.fill_rule=='evenodd'
|
|
|
|
|
|
def test_pcb_storage_keys_do_not_allow_traversal_or_arbitrary_formats(client):
|
|
for name in ['input.PcbDoc','old.PcbDoc','new.PcbDoc','pcb-comparison.json','attempt-1/old.svg']:
|
|
assert storage().path('a'*48+'/'+name).name==name.split('/')[-1]
|
|
for name in ['../input.PcbDoc','x.PcbDoc','input.PcbDoc.exe','pcb-comparison.json/extra','attempt-3/drawing.svg']:
|
|
with pytest.raises(ValueError):storage().path('a'*48+'/'+name)
|
|
|
|
|
|
def test_comparison_accepts_each_file_limit_and_invalid_side(client,monkeypatch):
|
|
from app.config import settings
|
|
monkeypatch.setenv('MAX_UPLOAD_BYTES','20000');settings.cache_clear()
|
|
# Each is 18 KiB; combined exceeds one per-file limit and is nevertheless a valid pair.
|
|
raw=pcb();assert 20000<len(raw)*2 and len(raw)<20000
|
|
created=create(client,raw,True).json()
|
|
identifier=created['id'];headers={'X-Viewer-Token':created['viewer_token']}
|
|
assert client.get(f'/api/v1/pcbs/comparisons/{identifier}/svg/other',headers=headers).status_code==422
|
|
assert client.get(f'/api/v1/pcbs/comparisons/{identifier}/svg/../old',headers=headers).status_code==404
|
|
assert create(client,raw+b'x'*20000).status_code==413
|
|
|
|
|
|
@pytest.mark.parametrize('prefix',['/view/pcb/','/compare/pcb/'])
|
|
@pytest.mark.parametrize('frame_option',['DENY','SAMEORIGIN'])
|
|
def test_pcb_preview_replaces_duplicate_csp_and_removes_xfo(client,monkeypatch,prefix,frame_option):
|
|
from fastapi.responses import Response
|
|
def page(*args,**kwargs):
|
|
response=Response('preview',media_type='text/html',headers={'X-Frame-Options':frame_option,'Content-Security-Policy':"frame-ancestors 'none'"})
|
|
response.headers.append('Content-Security-Policy',"frame-ancestors 'none'")
|
|
return response
|
|
monkeypatch.setattr('app.pcb_api.FileResponse',page)
|
|
response=client.get(prefix+'a'*48)
|
|
assert len(response.headers.get_list('content-security-policy'))==1
|
|
assert "frame-ancestors 'self' https://git.sadentel.net" in response.headers['content-security-policy']
|
|
assert 'x-frame-options' not in response.headers
|