103 lines
5.4 KiB
Python
103 lines
5.4 KiB
Python
"""Run the real private-import guard against a disposable database and fake Docker."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
from logistics_chat.order_data import import_records
|
|
|
|
|
|
@unittest.skipUnless(os.name == 'posix' and shutil.which('bash'), 'Private import requires Linux Bash')
|
|
class GrayOrderImportTests(unittest.TestCase):
|
|
def run_import(self, corrupt=False):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
directory = Path(temporary.name)
|
|
root = directory / 'gray-logistics-chat'
|
|
release_id = 'a' * 12 + '-b7'
|
|
release = root / 'releases' / release_id
|
|
release.mkdir(parents=True)
|
|
(root / '.project').write_text('baiyue-logistics-chat-gray\n')
|
|
(root / 'current-release').write_text(str(release) + '\n')
|
|
data = root / 'data'
|
|
(data / 'backups').mkdir(parents=True)
|
|
database = data / 'history.db'
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute('CREATE TABLE messages (content TEXT)')
|
|
connection.execute("INSERT INTO messages VALUES ('keep chat history')")
|
|
import_records(connection, [{'order_no': 'EXISTING1', 'source_order_id': '1',
|
|
'tenant_id': 'test', 'gross_weight_kg': 'manual weight',
|
|
'tracking_json': [{'manual': True}]}])
|
|
orders = [{'order_no': 'EXISTING1', 'source_order_id': '1', 'tenant_id': 'test', 'gross_weight_kg': 'source weight'},
|
|
{'order_no': 'NEW00002', 'source_order_id': '2', 'tenant_id': 'test'}]
|
|
tracking = [{'order_no': row['order_no'], 'source_order_id': row['source_order_id'],
|
|
'tenant_id': 'test', 'tracking_json': [{'code': 'IN_TRANSIT', 'actual_at': '2026-10-30'}]}
|
|
for row in orders]
|
|
hashes = []
|
|
for name, records in [('order-input.json', orders), ('tracking-input.json', tracking)]:
|
|
path = release / name
|
|
path.write_text(json.dumps({'orders': records}))
|
|
hashes.append(hashlib.sha256(path.read_bytes()).hexdigest())
|
|
if corrupt:
|
|
(release / 'order-input.json').write_text('changed in transit')
|
|
commands = directory / 'commands'
|
|
commands.mkdir()
|
|
docker = commands / 'docker'
|
|
docker.write_text('#!' + sys.executable + '\n' + '''import os, pathlib, subprocess, sys
|
|
args = sys.argv[1:]
|
|
if args[0] == 'inspect':
|
|
print('a'*40 if 'revision' in args[2] else 'baiyue-logistics-chat-gray')
|
|
elif args[0] == 'exec':
|
|
command = [sys.executable] + args[3:]
|
|
command = [value.replace('/var/lib/baiyue-logistics-chat', os.environ['FAKE_DATA']) for value in command]
|
|
if len(command) > 1 and command[1] != '-c':
|
|
command[1] = str(pathlib.Path(os.environ['FAKE_SOURCE']) / command[1])
|
|
sys.exit(subprocess.run(command).returncode)
|
|
else:
|
|
sys.exit(3)
|
|
''')
|
|
docker.chmod(0o755)
|
|
source = Path(__file__).resolve().parents[1]
|
|
script = (source / 'deploy/import-gray-order-samples.sh').read_text()
|
|
remote = script.split("<<'IMPORT'\n", 1)[1].split('\nIMPORT\n', 1)[0]
|
|
environment = dict(os.environ, PATH=str(commands) + os.pathsep + os.environ['PATH'],
|
|
FAKE_DATA=str(data), FAKE_SOURCE=str(source))
|
|
result = subprocess.run(['bash', '-s', '--', str(root), release_id, 'a' * 40, *hashes],
|
|
input=remote, env=environment, capture_output=True, text=True, timeout=30)
|
|
return result, data, release, release_id
|
|
|
|
def test_import_backs_up_and_preserves_manual_fields_tracking_and_chat_history(self):
|
|
result, data, release, release_id = self.run_import()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
with sqlite3.connect(data / 'history.db') as connection:
|
|
self.assertEqual(connection.execute('SELECT content FROM messages').fetchall(), [('keep chat history',)])
|
|
row = connection.execute('SELECT gross_weight_kg,tracking_json FROM order_test_data WHERE order_no="EXISTING1"').fetchone()
|
|
self.assertEqual(row[0], 'manual weight')
|
|
self.assertEqual(json.loads(row[1]), [{'manual': True}])
|
|
self.assertEqual(connection.execute('SELECT COUNT(*) FROM order_test_data').fetchone()[0], 2)
|
|
row = connection.execute('SELECT tracking_json FROM order_test_data WHERE order_no="NEW00002"').fetchone()
|
|
self.assertEqual(json.loads(row[0])[0]['code'], 'IN_TRANSIT')
|
|
with sqlite3.connect(data / 'backups' / ('pre-order-import-' + release_id + '.db')) as backup:
|
|
self.assertEqual(backup.execute('SELECT COUNT(*) FROM order_test_data').fetchone()[0], 1)
|
|
self.assertFalse((data / 'imports' / release_id).exists())
|
|
self.assertFalse((release / 'order-input.json').exists())
|
|
self.assertFalse((release / 'tracking-input.json').exists())
|
|
|
|
def test_corrupt_snapshot_is_rejected_before_database_changes(self):
|
|
result, data, release, release_id = self.run_import(corrupt=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
with sqlite3.connect(data / 'history.db') as connection:
|
|
self.assertEqual(connection.execute('SELECT COUNT(*) FROM order_test_data').fetchone()[0], 1)
|
|
self.assertFalse(list((data / 'backups').iterdir()))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|