42 lines
1.8 KiB
Python
42 lines
1.8 KiB
Python
import bz2
|
|
import io
|
|
import json
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
import sys
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
|
|
from sync import parse_archive,reconcile
|
|
from validate import validate
|
|
|
|
class SyncTests(unittest.TestCase):
|
|
def test_roundtrip_and_deletions(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
p=Path(tmp);archive=p/'test.tar.bz2'
|
|
data=json.dumps({'document':{'category':'csaf_vex'},'vulnerabilities':[]}).encode()
|
|
with tarfile.open(archive,'w:bz2') as tf:
|
|
for name in ('csaf-vex/2025/cve-2025-12345.json','csaf-vex/2016/cve-2016-9999.json'):
|
|
info=tarfile.TarInfo(name);info.size=len(data);tf.addfile(info,io.BytesIO(data))
|
|
items=parse_archive(archive,2017)
|
|
self.assertEqual(list(items),['2025/CVE-2025-12345.json'])
|
|
import os
|
|
cwd=Path.cwd();os.chdir(p)
|
|
try:
|
|
result=reconcile(items,Path('data/vex'),Path('state/sync.json'),'test',2017)
|
|
self.assertEqual(result['record_count'],1)
|
|
validate()
|
|
(Path('data/vex/2025')/'CVE-2025-12345.json').unlink()
|
|
reconcile(items,Path('data/vex'),Path('state/sync.json'),'test',2017)
|
|
validate()
|
|
finally:os.chdir(cwd)
|
|
def test_invalid_csaf_fails_before_write(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
p=Path(tmp);archive=p/'bad.tar.bz2'
|
|
data=b'{bad'
|
|
with tarfile.open(archive,'w:bz2') as tf:
|
|
x=tarfile.TarInfo('cve-2026-1111.json');x.size=len(data);tf.addfile(x,io.BytesIO(data))
|
|
with self.assertRaises(json.JSONDecodeError):parse_archive(archive,2017)
|
|
|
|
if __name__=='__main__':unittest.main()
|