import copy import json import re import time from pathlib import Path from xml.etree.ElementTree import fromstring import pytest from sqlalchemy import update from app.diff import compare_documents from app.parser.schdoc_parser import parse_schdoc from app.render import svg_document from app.db import Job,claim,sessions from app.storage import storage from app.worker import process from test_platform import client,KEY def parse(raw,name='test.SchDoc'):return parse_schdoc(raw,name,'test') def compare(a,b):return compare_documents(parse(a),parse(b)) def create(client,old,new,**fields): return client.post('/api/compare',headers={'X-API-Key':KEY}, files={'old_file':('old.SchDoc',old),'new_file':('new.SchDoc',new)},data=fields) def test_identical_and_record_reordering(ascii_schematic): assert compare(ascii_schematic,ascii_schematic)['total_changed']==0 lines=ascii_schematic.decode().splitlines();records=lines[1:] order=list(reversed(range(len(records))));mapping={old:new for new,old in enumerate(order)} shuffled=[re.sub(r'(?i)OwnerIndex=(\d+)',lambda m:f'OwnerIndex={mapping[int(m[1])]}',records[i]) for i in order] result=compare(ascii_schematic,(lines[0]+'\n'+'\n'.join(shuffled)).encode()) assert result['total_changed']==0,result['changes'] def test_component_properties_and_renumbering_with_native_identity(ascii_schematic): old=ascii_schematic.replace(b'LibReference=R_Symbol',b'UniqueId=COMPONENT01|LibReference=R_Symbol') new=old.replace(b'Text=R1',b'Text=R9').replace(b'Text=4K7',b'Text=10K') result=compare(old,new) assert result['summary']=={'added':0,'removed':0,'modified':1,'moved':0} change=result['changes'][0] assert change['kind']=='component' and change['match']=='unique_id' assert {'designator','value','parameters'}<={f['property'] for f in change['fields']} def move_component(raw,dx=20,dy=10): lines=[] for line in raw.decode().splitlines(): if '|RECORD=1|' in line or '|OwnerIndex=1|' in line: line=re.sub(r'(Location|Corner)\.([XY])=(-?\d+)',lambda m:f'{m[1]}.{m[2]}={int(m[3])+(dx if m[2]=="X" else dy)}',line) lines.append(line) return ('\n'.join(lines)+'\n').encode() def test_movement_without_inventing_internal_symbol_changes(ascii_schematic): result=compare(ascii_schematic,move_component(ascii_schematic)) assert result['summary']=={'added':0,'removed':0,'modified':0,'moved':1},result['changes'] assert result['changes'][0]['movement']=={'dx':20,'dy':-10} assert not result['changes'][0]['fields'] both=compare(ascii_schematic,move_component(ascii_schematic).replace(b'Text=4K7',b'Text=10K')) assert both['changes'][0]['moved'] and both['changes'][0]['status']=='modified' def test_added_removed_component_and_wire(ascii_schematic): added=ascii_schematic+b'''|RECORD=1|LibReference=C|UniqueId=CAP01|Location.X=300|Location.Y=400|CurrentPartId=1 |RECORD=13|OwnerIndex=9|Location.X=300|Location.Y=400|Corner.X=310|Corner.Y=410 |RECORD=34|OwnerIndex=9|Text=C2|Location.X=300|Location.Y=420 |RECORD=27|UniqueId=WIRE_NEW|LocationCount=2|X1=300|Y1=400|X2=350|Y2=400 ''' result=compare(ascii_schematic,added) assert result['summary']['added']==2 and result['total_changed']==2 assert {c['kind'] for c in result['changes']}=={'component','wire'} assert compare(added,ascii_schematic)['summary']['removed']==2 def test_nets_report_labels_not_compiled_connectivity(ascii_schematic): result=compare(ascii_schematic,ascii_schematic.replace(b'Text=VCC',b'Text=VBAT')) assert result['summary']['modified']==1 assert result['nets']=={'added':['VBAT'],'removed':['VCC'],'changed':[]} assert result['changes'][0]['kind']=='net_label' assert any('compiled' in message for message in result['limitations']) def test_wire_reversal_is_not_an_electrical_or_visual_change(ascii_schematic): new=ascii_schematic.replace(b'X1=100|Y1=230|X2=150|Y2=230',b'X1=150|Y1=230|X2=100|Y2=230') assert compare(ascii_schematic,new)['total_changed']==0 def test_duplicates_are_not_arbitrary_matches(ascii_schematic): extra=b'|RECORD=27|LocationCount=2|X1=10|Y1=10|X2=20|Y2=10\n' assert compare(ascii_schematic+extra*2,ascii_schematic+extra*2)['total_changed']==0 result=compare(ascii_schematic+extra*2,ascii_schematic+extra) assert result['summary']['removed']==1 and result['total_changed']==1 def test_report_does_not_mutate_documents_and_embedded_image_changes(ascii_schematic): from app.models.schematic import Primitive,Bounds,Point old=parse(ascii_schematic);new=old.model_copy(deep=True) before=old.model_dump_json();compare_documents(old,new);assert old.model_dump_json()==before # Existing parsed image record with a changed binary raster, same filename. obj=new.objects[-1].model_copy(deep=True);obj.id='image';obj.kind='image';obj.record_type=30 obj.properties={'RECORD':'30','FILENAME':'logo.bmp','LOCATION.X':'0','LOCATION.Y':'0'};obj.primitive_ids=['image:p'];obj.component_id=None old.objects.append(obj);new.objects.append(obj.model_copy(deep=True)) p=Primitive(id='image:p',kind='image',object_id='image',bounds=Bounds(x=0,y=0,width=10,height=10),image_data='data:image/png;base64,OLD') old.primitives.append(p);new.primitives.append(p.model_copy(update={'image_data':'data:image/png;base64,NEW'})) assert compare_documents(old,new)['summary']['modified']==1 def test_highlight_svg_preserves_geometry_and_distinct_clip_ids(ascii_schematic): raw=ascii_schematic+b'|RECORD=28|Location.X=1|Corner.X=20|Corner.Y=20|ClipToRect=T|Text=note\n' doc=parse(raw);item=doc.objects[-1] annotations={item.id:{'id':'change-0','status':'modified','bounds':item.bounds.model_dump()}} svg=svg_document(doc,annotations,id_prefix='old-');tree=fromstring(svg) assert b'data-diff-box="true"' in svg and b'data-change-id="change-0"' in svg assert b'old-clip-' in svg and b'url(#old-clip-' in svg assert b'M100,-230 L150,-230' in svg assert not list(tree.iter('{http://www.w3.org/2000/svg}script')) def test_comparison_api_worker_private_results_and_cleanup(client,ascii_schematic): response=create(client,ascii_schematic,ascii_schematic.replace(b'Text=4K7',b'Text=10K')) assert response.status_code==202,response.text created=response.json();identifier=created['comparison_id'];headers={'X-Viewer-Token':created['viewer_token']} assert '#token=' in created['url'] and created['status']=='queued' base='/api/v1/comparisons/'+identifier assert client.get(base).status_code==404 assert client.get(base,headers={'X-Viewer-Token':'bad'}).status_code==404 assert client.get(base+'/svg/old',headers=headers).status_code==409 job=claim();process(job) response=client.get(base,headers=headers);result=response.json() assert result['status']=='completed',result assert result['summary']['modified']==1 assert client.get('/api/compare/'+identifier,headers=headers).json()['changes']==result['changes'] for side in ('old','new'): r=client.get(base+'/svg/'+side,headers=headers) assert r.status_code==200 and b'data-diff-box' in r.content assert client.get('/compare/'+identifier).status_code==200 assert client.get('/compare').status_code==200 for asset in ['compare.js','compare.css','interactions.js','compare-upload.js']: assert client.get('/assets/'+asset).status_code==200 with sessions().begin() as db:db.execute(update(Job).where(Job.id==identifier).values(expires_at=time.time()-1)) assert client.get(base,headers=headers).status_code==410 from app.cleanup import cleanup assert cleanup()==1 with pytest.raises(FileNotFoundError):storage().get(identifier+'/old.SchDoc') def test_comparison_validation_auth_and_queue_limit(client,ascii_schematic,monkeypatch): assert client.post('/api/compare').status_code==401 assert client.post('/api/v1/comparisons').status_code==401 assert create(client,ascii_schematic,b'garbage').status_code==422 assert create(client,ascii_schematic,ascii_schematic,expires_in=999999999).status_code==422 assert create(client,b'x'*(1048576+1),ascii_schematic).status_code==413 monkeypatch.setenv('MAX_PENDING_JOBS','1') from app.config import settings settings.cache_clear() assert create(client,ascii_schematic,ascii_schematic).status_code==202 assert create(client,ascii_schematic,ascii_schematic).status_code==503 def test_corrupt_second_file_fails_durably_and_unlisted(client,ascii_schematic): response=create(client,ascii_schematic,b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1'+b'\0'*504,access='unlisted') assert response.status_code==202 process(claim()) result=client.get('/api/compare/'+response.json()['id']).json() assert result['status']=='failed' and result['error']['code']=='invalid_schematic' def test_comparison_cross_origin_preflight_and_creation(client,ascii_schematic): origin='https://git.sadentel.net' r=client.options('/api/compare',headers={'Origin':origin,'Access-Control-Request-Method':'POST','Access-Control-Request-Headers':'x-api-key'}) assert r.status_code==200 and r.headers['access-control-allow-origin']==origin r=client.post('/api/compare',headers={'Origin':origin}) assert r.status_code==401 and r.headers['access-control-allow-origin']==origin def test_partial_storage_failure_leaves_no_comparison_job(client,ascii_schematic,monkeypatch): original=storage().put;calls=[] def failing(key,body,*args,**kw): calls.append(key) if len(calls)==2:raise RuntimeError('S3 unavailable') return original(key,body,*args,**kw) monkeypatch.setattr(storage(),'put',failing) assert create(client,ascii_schematic,ascii_schematic).status_code==503 assert claim() is None assert not storage().path(calls[0]).exists() def test_native_ole_comparison_preserves_original_geometry(client): fixture=Path(__file__).resolve().parents[2]/'.runtime/fixtures/Channel.SchDoc' if not fixture.is_file():pytest.skip('Fetch the native fixture') response=create(client,fixture.read_bytes(),fixture.read_bytes());assert response.status_code==202 process(claim()) identifier=response.json()['id'];headers={'X-Viewer-Token':response.json()['viewer_token']} result=client.get('/api/v1/comparisons/'+identifier,headers=headers).json() assert result['status']=='completed',result assert result['total_changed']==0 assert result['old']['sha256']==result['new']['sha256'] assert b'data-object-id' in client.get(f'/api/v1/comparisons/{identifier}/svg/old',headers=headers).content def test_comparison_request_body_is_bounded(client): response=client.post('/api/compare',content=b'x'*(4*1048576), headers={'X-API-Key':KEY,'Content-Type':'multipart/form-data; boundary=foo'}) assert response.status_code==413 def test_comparison_with_valid_empty_sheet_keeps_single_parser_behavior(client,ascii_schematic): from app.parser.record_decoder import ParseError empty=b'|HEADER=Protel for Windows - Schematic Capture Ascii File Version 5.0\n|RECORD=31|FontIdCount=1|Size1=10|FontName1=Arial\n' with pytest.raises(ParseError,match='No supported visible geometry'):parse(empty) response=create(client,ascii_schematic,empty);assert response.status_code==202 process(claim());headers={'X-Viewer-Token':response.json()['viewer_token']} result=client.get('/api/compare/'+response.json()['id'],headers=headers).json() assert result['status']=='completed',result assert result['summary']['removed']==3 assert result['summary']['added']==0 svg=client.get('/api/v1/comparisons/'+response.json()['id']+'/svg/new',headers=headers) assert svg.status_code==200 and b'