اضافه شدن قابلیت رندر پی سی بی

backend/tests/test_pcb.py

@@ -0,0 +1,298 @@
+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