228 خطوط
12 KiB
Python
228 خطوط
12 KiB
Python
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'<rect' in svg.content
|
|
|
|
|
|
def test_comparison_ids_do_not_become_single_schematic_resources(client,ascii_schematic):
|
|
response=create(client,ascii_schematic,ascii_schematic)
|
|
identifier=response.json()['id'];headers={'X-Viewer-Token':response.json()['viewer_token']}
|
|
for endpoint in ('status','data','svg'):
|
|
assert client.get(f'/api/v1/schematics/{identifier}/{endpoint}',headers=headers).status_code==404
|
|
from test_platform import upload
|
|
single=upload(client,ascii_schematic).json()
|
|
assert client.get('/api/compare/'+single['id'],headers={'X-Viewer-Token':single['viewer_token']}).status_code==404
|