tests: Add state dumping into RepositorySimulator

if state dumping is enabled with e.g.
    python3 test_updater_with_simulator.py --dump
The repository state can be dumped at will.

Modify the test so it dumps the state on every updater refresh if
--dump is set.

Add a root modifying case to test_refresh()

Signed-off-by: Jussi Kukkonen <jkukkonen@vmware.com>

Author: Jussi Kukkonen

tests/repository_simulator.py

@@ -15,16 +15,18 @@
 everything from memory.
 """
 
-import logging
 from collections import OrderedDict
 from datetime import datetime, timedelta
-from tuf.api.serialization.json import JSONSerializer
+import logging
+import os
+import tempfile
 from securesystemslib.keys import generate_ed25519_key
 from securesystemslib.signer import SSlibSigner
-from tuf.exceptions import FetcherHTTPError
 from typing import Dict, Iterator, List, Optional, Tuple
 from urllib import parse
 
+from tuf.api.serialization.json import JSONSerializer
+from tuf.exceptions import FetcherHTTPError
 from tuf.api.metadata import(
     Key,
     Metadata,
@@ -57,6 +59,9 @@ def __init__(self):
         # signers are used on-demand at fetch time to sign metadata
         self.signers: Dict[str, List[SSlibSigner]] = {}
 
+        self.dump_dir = None
+        self.dump_version = 0
+
         self._initialize()
 
     @property
@@ -179,3 +184,29 @@ def update_snapshot(self):
 
         self.snapshot.version += 1
         self.update_timestamp()
+
+    def write(self):
+        """Dump current repository metadata to self.dump_dir
+
+        This is a debugging tool: dumping repository state before running
+        Updater refresh may be useful while debugging a test.
+        """
+        if self.dump_dir is None:
+            self.dump_dir = tempfile.mkdtemp()
+            print(f"Repository Simulator dumps in {self.dump_dir}")
+
+        self.dump_version += 1
+        dir = os.path.join(self.dump_dir, str(self.dump_version))
+        os.makedirs(dir)
+
+        for ver in range(1, len(self.signed_roots) + 1):
+            with open(os.path.join(dir, f"{ver}.root.json"), "wb") as f:
+                f.write(self._fetch_metadata("root", ver))
+
+        for role in ["timestamp", "snapshot", "targets"]:
+            with open(os.path.join(dir, f"{role}.json"), "wb") as f:
+                f.write(self._fetch_metadata(role))
+
+        for role in self.md_delegates.keys():
+            with open(os.path.join(dir, f"{role}.json"), "wb") as f:
+                f.write(self._fetch_metadata(role))

tests/test_updater_with_simulator.py

@@ -19,6 +19,9 @@
 from tests.repository_simulator import RepositorySimulator
 
 class TestUpdater(unittest.TestCase):
+    # set dump_dir to trigger repository state dumps
+    dump_dir:str = None
+
     def setUp(self):
         self.client_dir = tempfile.TemporaryDirectory()
 
@@ -28,81 +31,95 @@ def setUp(self):
             root = self.sim.download_bytes("https://example.com/metadata/1.root.json", 100000)
             f.write(root)
 
-    def _new_updater(self):
-        return Updater(
+        if self.dump_dir is not None:
+            # create test specific dump directory
+            name = self.id().split('.')[-1]
+            self.sim.dump_dir = os.path.join(self.dump_dir, name)
+            os.mkdir(self.sim.dump_dir)
+
+    def _run_refresh(self):
+        if self.sim.dump_dir is not None:
+            self.sim.write()
+
+        updater = Updater(
             self.client_dir.name,
             "https://example.com/metadata/",
             "https://example.com/targets/",
             self.sim
         )
+        updater.refresh()
 
     def test_refresh(self):
+        # TODO: how do we ensure updater does what we expected it to?
+        # TODO: how do we ensure updater state is correct after update()?
+
         # Update top level metadata
-        self._new_updater().refresh()
+        self._run_refresh()
 
-        # New root (root needs to be explicitly published)
+        # New root (root needs to be explicitly signed)
         self.sim.root.version += 1
         self.sim.publish_root()
 
-        # TODO compare file contents?
+        self._run_refresh()
 
-        # New timestamp version
+        # New timestamp
         self.sim.update_timestamp()
 
-        self._new_updater().refresh()
+        self._run_refresh()
 
-        # TODO compare file contents?
-
-        # New targets version
+        # New targets, snapshot, timestamp version
         self.sim.targets.version += 1
         self.sim.update_snapshot()
 
-        self._new_updater().refresh()
+        self._run_refresh()
 
-        # TODO compare file contents?
+    def test_keys_and_signatures(self):
+        """Example of the two trickiest test areas: keys and root updates"""
 
-    # this is just an example of testing different key/signature situations
-    def test_targets_signatures(self):
         # Update top level metadata
-        self._new_updater().refresh()
+        self._run_refresh()
 
-        # New targets: signed by a new key that is not in roles keys
+        # New targets: signed with a new key that is not in roles keys
         old_signer = self.sim.signers["targets"].pop()
         key, signer = self.sim.create_key()
         self.sim.signers["targets"] = [signer]
         self.sim.targets.version += 1
         self.sim.update_snapshot()
 
         with self.assertRaises(UnsignedMetadataError):
-            self._new_updater().refresh()
+            self._run_refresh()
 
         # New root: Add the new key as targets role key
         # (root changes require explicit publishing)
         self.sim.root.add_key("targets", key)
         self.sim.root.version += 1
         self.sim.publish_root()
 
-        self._new_updater().refresh()
+        self._run_refresh()
 
         # New root: Raise targets threshold to 2
         self.sim.root.roles["targets"].threshold = 2
         self.sim.root.version += 1
         self.sim.publish_root()
 
         with self.assertRaises(UnsignedMetadataError):
-            self._new_updater().refresh()
+            self._run_refresh()
 
         # New targets: sign with both new and old key
         self.sim.signers["targets"] = [signer, old_signer]
         self.sim.targets.version += 1
         self.sim.update_snapshot()
 
-        self._new_updater().refresh()
-
+        self._run_refresh()
 
     def tearDown(self):
         self.client_dir.cleanup()
 
 if __name__ == "__main__":
-  utils.configure_test_logging(sys.argv)
-  unittest.main()
+    if "--dump" in sys.argv:
+        TestUpdater.dump_dir = tempfile.mkdtemp()
+        print(f"Repository Simulator dumps in {TestUpdater.dump_dir}")
+        sys.argv.remove("--dump")
+
+    utils.configure_test_logging(sys.argv)
+    unittest.main()