38 lines
1.8 KiB
Python
38 lines
1.8 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from scripts.sync import process, validate_page, save_atomic, stamp
|
|
import datetime as dt
|
|
|
|
A = "12345678-1234-1234-1234-123456789abc"
|
|
B = "87654321-4321-4321-4321-cba987654321"
|
|
|
|
class SyncTest(unittest.TestCase):
|
|
def test_validation(self):
|
|
page = {"startIndex":0,"totalResults":1,"products":[{"cpe":{"cpeNameId":A,"cpeName":"cpe:2.3:a:example:test:1:*:*:*:*:*:*:*"}}]}
|
|
self.assertEqual(len(validate_page(page,"products","cpe","cpeNameId")),1)
|
|
with self.assertRaises(ValueError):
|
|
validate_page({"startIndex":0,"totalResults":1,"products":[{}]},"products","cpe","cpeNameId")
|
|
def test_resume_full_then_delta(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
root = Path(t)
|
|
def fake_fetch(url, params, key, delay):
|
|
if "lastModStartDate" in params: return {"startIndex":0,"totalResults":0,"products":[]}
|
|
index=params["startIndex"]
|
|
return {"startIndex":index,"totalResults":2,"products":[{"cpe":{"cpeNameId": [A,B][index]}}]}
|
|
with patch("scripts.sync.fetch", fake_fetch):
|
|
processed,_,mode=process("cpes",root,"",1,0)
|
|
self.assertEqual((processed,mode),(1,"full"))
|
|
processed,_,mode=process("cpes",root,"",1,0)
|
|
self.assertEqual((processed,mode),(1,"delta"))
|
|
self.assertEqual(len(list((root/"data/cpes").rglob("*.json"))),2)
|
|
def test_atomic_equal(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
p = Path(t)/"x.json"
|
|
self.assertTrue(save_atomic(p,{"a":1}))
|
|
self.assertFalse(save_atomic(p,{"a":1}))
|
|
|
|
if __name__ == "__main__": unittest.main()
|