44 lines
2.3 KiB
Python
44 lines
2.3 KiB
Python
|
import hashlib
|
||
|
|
from pathlib import Path
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from console.settings import Settings
|
||
|
|
from console.store import Store
|
||
|
|
from console.catalog import discover
|
||
|
|
from deploy.refresh_catalog import refresh, restore
|
||
|
|
|
||
|
|
|
||
|
|
class CatalogRefreshTests(unittest.TestCase):
|
||
|
|
def test_refresh_preserves_options_and_history_and_restores_exactly(self):
|
||
|
|
with tempfile.TemporaryDirectory() as temporary:
|
||
|
|
base = Path(temporary); source = base/'source'; source.mkdir()
|
||
|
|
(source/'old.py').write_text('print(1)')
|
||
|
|
settings = Settings(base/'runtime', [{'id':'suite','path':str(source)}]); store = Store(settings)
|
||
|
|
discover(store)
|
||
|
|
old = store.scripts()[0]; old.update(enabled=True,status='ready',reason='')
|
||
|
|
old['config']['timeout_seconds'] = 457
|
||
|
|
store.save_script(old)
|
||
|
|
old_runs = store.runs()
|
||
|
|
(source/'old.py').write_text('print(5)')
|
||
|
|
(source/'new.py').write_text('print(2)'); (source/'blocked.py').write_text('print(3)')
|
||
|
|
(source/'support.py').write_text('print(4)')
|
||
|
|
settings.native_catalog['profiles'] = [
|
||
|
|
{'id':name,'source_group':'suite','entry':name+'.py','cwd':str(source),'interpreter':'/usr/bin/python3',
|
||
|
|
'user':'root','available':name!='blocked','unavailable_reason':'原服务预设已过期','parameters':[]}
|
||
|
|
for name in ('old','new','blocked')]
|
||
|
|
snapshot = base/'migration.json'
|
||
|
|
self.assertFalse(refresh(store, snapshot)['applied'])
|
||
|
|
self.assertFalse(snapshot.exists())
|
||
|
|
refresh(store, snapshot, True)
|
||
|
|
by_name = {s['name']:s for s in store.scripts()}
|
||
|
|
self.assertEqual(by_name['old.py']['config']['timeout_seconds'],457)
|
||
|
|
self.assertFalse(by_name['old.py']['enabled'])
|
||
|
|
self.assertEqual(by_name['old.py']['version'],2)
|
||
|
|
self.assertTrue(by_name['new.py']['enabled'])
|
||
|
|
self.assertFalse(by_name['blocked.py']['enabled'])
|
||
|
|
self.assertFalse(by_name['support.py']['enabled'])
|
||
|
|
self.assertEqual(store.runs(),old_runs)
|
||
|
|
self.assertEqual(refresh(store,base/'again.json',True)['changed'],0)
|
||
|
|
restore(store,snapshot)
|
||
|
|
self.assertEqual(store.scripts(),[old])
|