Harden OCI catalog refresh recovery and retries

oci/CATALOG-REFRESH.md

@@ -6,6 +6,10 @@ catalog, and changed recipes. Returning from a search or installation does not
 start another download. A running wizard keeps the catalog it started with.
 The menu also offers **Refresh catalog** to check again manually or retry after
 a connection failure. Selecting it during a refresh does not start a second worker.
+Manual refresh bypasses the conditional check and verifies the complete manifest,
+allowing it to repair damaged cached files even when the upstream revision has not
+changed. A repaired snapshot gets a separate immutable directory so existing
+readers retain their original snapshot until they return to the menu.
 
 The worker checks the latest commit affecting `oci/catalog` in the official
 MacRimi/ProxMenux repository. GitHub's ETag avoids repeated transfers. When the
@@ -20,6 +24,8 @@ the engine replaced by the normal ProxMenux updater. The bundled catalog remains
 the fallback. A network failure, incomplete manifest, invalid recipe, different
 template schema or unsupported catalog engine API leaves the previous catalog
 active. The private `CHANNEL-result.json` records the diagnostic.
+The revision, snapshot and ETag are committed together in one atomic pointer;
+failure to publish it cannot acknowledge an update that was not applied.
 
 The channel follows `install_branch`, an active beta program or `beta_version.txt`;
 otherwise it is `main`. A host can explicitly select `main` or `develop` with

oci/src/proxmenux_oci/catalog_refresh.py

@@ -14,12 +14,15 @@ import json
 import os
 from pathlib import Path, PurePosixPath
 import re
+import shutil
 import subprocess
 import sys
 import tempfile
 import threading
+import time
 import urllib.error
 import urllib.request
+import uuid
 
 from .i18n import translate
 
@@ -46,11 +49,12 @@ def channel_for(root):
             root.parent if root.name == "oci" else root)
     for name in ("oci/catalog-source.json", "config.json"):
         try:
-            value = json.loads((base / name).read_text())
+            value = read_json(base / name)
             channel = value.get("channel") or value.get("install_branch")
             if channel in ("main", "develop"):
                 return channel
-            if (value.get("beta_program") or {}).get("status") == "active":
+            beta = value.get("beta_program")
+            if isinstance(beta, dict) and beta.get("status") == "active":
                 return "develop"
         except (OSError, ValueError, TypeError):
             pass
@@ -99,7 +103,10 @@ def current_catalog(root, channel=None):
     if (isinstance(revision, str) and SHA.fullmatch(revision)
             and value.get("schema_hash") == schema_hash(root)
             and value.get("engine_api") == CATALOG_API):
-        path = cache_root(root) / "snapshots" / revision
+        snapshot = value.get("snapshot", revision)
+        if not isinstance(snapshot, str) or not re.fullmatch(revision + r"(?:-[0-9a-f]{12})?", snapshot):
+            return root / "catalog"
+        path = cache_root(root) / "snapshots" / snapshot
         index = read_json(path / "index.json")
         if (index.get("kind") == "proxmenux.oci-catalog-index"
                 and isinstance(index.get("applications"), list) and index["applications"]):
@@ -122,6 +129,7 @@ def request(url, etag=None, limit=MAX_FILE):
                 raise ValueError("Catalog response exceeds the size limit")
             return body, response.headers.get("ETag")
     except urllib.error.HTTPError as error:
+        error.close()
         if error.code == 304:
             return None, etag
         raise
@@ -171,15 +179,15 @@ def validate_snapshot(root, directory):
             validator.validate(catalog._compose_item(item))
 
 
-def synchronize(root, channel, fetch=request):
+def synchronize(root, channel, fetch=request, force=False):
     """Download changed JSON files into a private snapshot, then publish once."""
     root = Path(root)
     cache = cache_root(root)
     cache.mkdir(parents=True, exist_ok=True)
-    metadata_path = cache / f"{channel}-remote.json"
-    previous = read_json(metadata_path)
+    pointer = cache / f"{channel}-current.json"
+    previous = read_json(pointer)
     active = current_catalog(root, channel)
-    etag = previous.get("etag") if active != root / "catalog" else None
+    etag = previous.get("etag") if active != root / "catalog" and not force else None
     url = f"https://api.github.com/repos/{REPOSITORY}/commits?sha={channel}&path=oci/catalog&per_page=1"
     body, etag = fetch(url, etag)
     if body is None:
@@ -188,8 +196,8 @@ def synchronize(root, channel, fetch=request):
     revision = commits[0]["sha"]
     if not isinstance(revision, str) or not SHA.fullmatch(revision):
         raise ValueError("Invalid catalog revision")
-    if active.name == revision:
-        write_json(metadata_path, {"etag": etag})
+    if previous.get("revision") == revision and active != root / "catalog" and not force:
+        write_json(pointer, {**previous, "etag": etag})
         return False
     body, _ = fetch(f"https://api.github.com/repos/{REPOSITORY}/git/trees/{revision}?recursive=1")
     tree = json.loads(body)
@@ -211,6 +219,8 @@ def synchronize(root, channel, fetch=request):
             raise ValueError("Invalid catalog manifest entry")
         if type(entry.get("size")) is not int or not 0 <= entry["size"] <= MAX_FILE:
             raise ValueError("Invalid catalog file size")
+        if relative in entries:
+            raise ValueError("Duplicate catalog manifest path")
         entries[relative] = entry
     if schema != git_blob((root / "schemas/oci-template.schema.json").read_bytes()):
         raise ValueError("Catalog schema changed; update ProxMenux first")
@@ -222,8 +232,11 @@ def synchronize(root, channel, fetch=request):
     snapshots.mkdir(exist_ok=True)
     with tempfile.TemporaryDirectory(prefix=".download-", dir=snapshots) as name:
         staging = Path(name)
+        deadline = time.monotonic() + 90
 
         def download(pair):
+            if time.monotonic() > deadline:
+                raise TimeoutError("Catalog download exceeded its time limit")
             relative, entry = pair
             old = active / relative
             data = old.read_bytes() if old.is_file() else b""
@@ -236,28 +249,40 @@ def synchronize(root, channel, fetch=request):
             target.parent.mkdir(parents=True, exist_ok=True)
             target.write_bytes(data)
 
-        with ThreadPoolExecutor(max_workers=4) as pool:
+        pool = ThreadPoolExecutor(max_workers=4)
+        try:
             list(pool.map(download, entries.items()))
+        finally:
+            pool.shutdown(wait=True, cancel_futures=True)
         validate_snapshot(root, staging)
-        destination = snapshots / revision
+        destination = (active if previous.get("revision") == revision and active != root / "catalog"
+                       else snapshots / revision)
         if destination.exists():
-            # Revalidate an existing immutable snapshot after an interrupted publication.
-            validate_snapshot(root, destination)
-        else:
+            intact = all((destination / relative).is_file()
+                         and not (destination / relative).is_symlink()
+                         and git_blob((destination / relative).read_bytes()) == entry["sha"]
+                         for relative, entry in entries.items())
+            existing = {str(p.relative_to(destination)) for p in destination.rglob("*.json")}
+            if not intact or existing != set(entries):
+                # Repair without removing a snapshot another wizard may still hold.
+                destination = snapshots / (revision + "-" + uuid.uuid4().hex[:12])
+        if not destination.exists():
             os.replace(staging, destination)
-        write_json(cache / f"{channel}-current.json", {"revision": revision,
+        write_json(pointer, {"revision": revision, "snapshot": destination.name, "etag": etag,
                    "schema_hash": schema_hash(root), "engine_api": CATALOG_API})
-        write_json(metadata_path, {"etag": etag})
-    return True
+    return destination != active
 
 
-def worker(root, channel):
+def worker(root, channel, force=False):
     cache = cache_root(root)
     cache.mkdir(parents=True, exist_ok=True)
     with (cache / "refresh.lock").open("a") as lock:
         fcntl.flock(lock, fcntl.LOCK_EX)
         try:
-            changed = synchronize(root, channel)
+            for partial in (cache / "snapshots").glob(".download-*"):
+                if partial.is_dir() and not partial.is_symlink():
+                    shutil.rmtree(partial)
+            changed = synchronize(root, channel, force=force)
             write_json(cache / f"{channel}-result.json", {"ok": True, "changed": changed})
         except Exception as error:
             write_json(cache / f"{channel}-result.json", {"ok": False, "error": str(error)[:400]})
@@ -300,6 +325,7 @@ class CatalogRefresh:
         self.channel = channel_for(root)
         self.seen_path = cache_root(root) / f"{self.channel}-seen.json"
         self.recipes = visible_recipes(catalog)
+        self.displayed_directory = catalog.catalog_dir
         baseline = read_json(self.seen_path).get("recipes")
         self.baseline = baseline if isinstance(baseline, dict) else self.recipes.copy()
         self.process = None
@@ -310,14 +336,17 @@ class CatalogRefresh:
         self.start()
         return self
 
-    def start(self):
+    def start(self, force=False):
         if self.active:
             return False
         self._finished = False
         self.failed = False
         environment = dict(os.environ, PYTHONPATH=str(self.root / "src"))
         try:
-            self.process = subprocess.Popen([sys.executable, "-m", __name__, str(self.root), self.channel],
+            command = [sys.executable, "-m", __name__, str(self.root), self.channel]
+            if force:
+                command.append("--force")
+            self.process = subprocess.Popen(command,
                                             env=environment, stdin=subprocess.DEVNULL,
                                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                             start_new_session=True)
@@ -348,7 +377,11 @@ class CatalogRefresh:
         self._finished = True
         result = read_json(cache_root(self.root) / f"{self.channel}-result.json")
         self.failed = self.process.returncode != 0 or not result.get("ok")
-        return not self.failed and bool(result.get("changed"))
+        directory = current_catalog(self.root, self.channel)
+        changed = directory != self.displayed_directory
+        if changed:
+            self.displayed_directory = directory
+        return changed
 
     def describe(self):
         text = translate("Available applications: {count}").format(count=len(self.recipes))
@@ -374,5 +407,6 @@ if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument("root", type=Path)
     parser.add_argument("channel", choices=("main", "develop"))
+    parser.add_argument("--force", action="store_true")
     args = parser.parse_args()
-    worker(args.root, args.channel)
+    worker(args.root, args.channel, args.force)

oci/src/proxmenux_oci/cli.py

@@ -626,7 +626,7 @@ def interactive(catalog: Catalog) -> int:
                     from .management import interactive_management
                     interactive_management(PROJECT_ROOT, ui)
                 elif selection == "catalog-refresh":
-                    catalog_refresh.start()
+                    catalog_refresh.start(force=True)
                 elif selection == "updates":
                     if not updates.active:
                         if state['available']:

oci/tests/test_catalog_refresh.py

@@ -1,10 +1,17 @@
 """Catalog delivery is atomic, offline-safe, and never acknowledges unseen apps."""
 import json
+import errno
+import http.server
+import itertools
 from pathlib import Path
 import shutil
 import sys
 import tempfile
+import threading
+import time
 import unittest
+import urllib.error
+import urllib.parse
 from unittest.mock import Mock, patch
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -303,6 +310,272 @@ class CatalogRefreshTests(unittest.TestCase):
         options = dict(_catalog_options(Catalog(self.root))[-1])
         self.assertEqual(options['catalog-refresh'], 'Refresh catalog')
 
+    def test_sequence_new_app_failed_update_retry_and_withdrawal(self):
+        self.sync()
+        old = Catalog(self.root)
+        status = refresh.CatalogRefresh(self.root, old)
+        status.mark_seen()
+        self.revision = 'b' * 40
+        self.add_app()
+        self.assertTrue(self.sync())
+        self.assertIn('sample-added', refresh.visible_recipes(Catalog(self.root)))
+        self.assertNotIn('sample-added', refresh.visible_recipes(old))  # Wizard remains pinned.
+        next_visit = refresh.CatalogRefresh(self.root, Catalog(self.root))
+        self.assertIn('New applications since your last visit: 1', next_visit.describe())
+        next_visit.mark_seen()
+        self.revision = 'c' * 40
+        recipe_path = self.item['curated_path'].removeprefix('catalog/')
+        good = self.files[recipe_path]
+        self.files[recipe_path] = b'{invalid-json'
+        with self.assertRaises(ValueError):
+            self.sync()
+        self.assertEqual(refresh.current_catalog(self.root).name, 'b' * 40)
+        self.files[recipe_path] = good
+        self.assertTrue(self.sync())
+        index = json.loads(self.files['index.json'])
+        index['applications'] = [self.item]
+        self.files['index.json'] = json.dumps(index).encode()
+        for name in ('apps/sample-added.json', 'curated/sample-added.json'):
+            self.files.pop(name)
+        self.revision = 'd' * 40
+        self.assertTrue(self.sync())
+        current = Catalog(self.root)
+        self.assertNotIn('sample-added', refresh.visible_recipes(current))
+        self.assertFalse((current.catalog_dir / 'curated/sample-added.json').exists())
+        self.assertIn('New applications since your last visit: 0',
+                      refresh.CatalogRefresh(self.root, current).describe())
+
+    def test_http_errors_and_timeouts_keep_snapshot_then_retry_succeeds(self):
+        self.sync()
+        old = refresh.current_catalog(self.root)
+        errors = [TimeoutError('timeout'), urllib.error.URLError('DNS unavailable')]
+        errors += [urllib.error.HTTPError('https://api.github.com', code, 'test', {}, None)
+                   for code in (403, 404, 429, 500, 503)]
+        for error in errors:
+            if isinstance(error, urllib.error.HTTPError): self.addCleanup(error.close)
+        self.revision = 'b' * 40
+        for error in errors:
+            with self.subTest(error=str(error)), self.assertRaises(Exception):
+                refresh.synchronize(self.root, 'main', Mock(side_effect=error))
+            self.assertEqual(refresh.current_catalog(self.root), old)
+        self.assertTrue(self.sync())
+
+    def test_failed_file_transfer_does_not_publish_partial_download(self):
+        self.sync(); old = refresh.current_catalog(self.root)
+        self.revision = 'b' * 40; self.add_app()
+        def interrupted(url, etag=None):
+            if 'raw.githubusercontent' in url:
+                raise TimeoutError('Connection lost during download')
+            return self.fetch(url, etag)
+        with self.assertRaises(TimeoutError):
+            refresh.synchronize(self.root, 'main', interrupted)
+        self.assertEqual(refresh.current_catalog(self.root), old)
+        self.assertFalse(list((refresh.cache_root(self.root) / 'snapshots').glob('.download-*')))
+        self.assertTrue(self.sync())
+
+    def test_disk_full_and_pointer_permission_failure_preserve_previous_revision_and_etag(self):
+        for error in (OSError(errno.ENOSPC, 'Disk full'), PermissionError('Read only')):
+            with self.subTest(error=error):
+                self.sync()
+                pointer = refresh.cache_root(self.root) / 'main-current.json'
+                before = pointer.read_bytes()
+                old = refresh.current_catalog(self.root)
+                self.revision = 'b' * 40
+                real = refresh.write_json
+                def fail(path, value):
+                    if Path(path) == pointer:
+                        raise error
+                    return real(path, value)
+                with patch.object(refresh, 'write_json', side_effect=fail), self.assertRaises(OSError):
+                    self.sync()
+                self.assertEqual(pointer.read_bytes(), before)
+                self.assertEqual(refresh.current_catalog(self.root), old)
+                self.assertTrue(self.sync())
+                self.revision = 'c' * 40
+
+    def test_manual_force_repairs_corrupted_recipe_at_same_revision(self):
+        self.sync()
+        old = refresh.current_catalog(self.root)
+        recipe = old / self.item['curated_path'].removeprefix('catalog/')
+        recipe.write_text('{}')
+        self.requests.clear()
+        self.assertTrue(refresh.synchronize(self.root, 'main', self.fetch, force=True))
+        self.assertIsNone(self.requests[0][1])
+        self.assertNotEqual(refresh.current_catalog(self.root), old)
+        self.assertEqual(Catalog(self.root).compose(self.item['id'])['id'], self.item['template_id'])
+        self.assertFalse(refresh.synchronize(self.root, 'main', self.fetch, force=True))
+
+    def test_corrupt_cached_index_can_be_repaired_by_automatic_refresh(self):
+        self.sync()
+        old = refresh.current_catalog(self.root)
+        (old / 'index.json').write_text('{broken')
+        self.assertTrue(self.sync())
+        self.assertNotEqual(refresh.current_catalog(self.root), self.root / 'catalog')
+        self.assertEqual(Catalog(self.root).compose(self.item['id'])['id'], self.item['template_id'])
+
+    def test_second_session_result_cannot_hide_successful_update(self):
+        status = refresh.CatalogRefresh(self.root, Catalog(self.root))
+        self.sync()
+        status.process = Mock(returncode=0)
+        status.process.poll.return_value = 0
+        # Another session checked the newly installed revision and found no change.
+        refresh.write_json(refresh.cache_root(self.root) / 'main-result.json',
+                           {'ok': True, 'changed': False})
+        self.assertTrue(status.consume())
+
+    def test_crashed_worker_does_not_trust_stale_success(self):
+        status = refresh.CatalogRefresh(self.root, Catalog(self.root))
+        status.process = Mock(returncode=1); status.process.poll.return_value = 1
+        refresh.write_json(refresh.cache_root(self.root) / 'main-result.json',
+                           {'ok': True, 'changed': True})
+        self.assertFalse(status.consume())
+        self.assertTrue(status.failed)
+
+    def test_force_manual_refresh_passes_flag_to_worker(self):
+        status = refresh.CatalogRefresh(self.root, Catalog(self.root))
+        with patch.object(refresh.subprocess, 'Popen') as spawn:
+            self.assertTrue(status.start(force=True))
+            self.assertIn('--force', spawn.call_args.args[0])
+
+    def test_duplicate_manifest_path_symlink_and_truncated_tree_are_rejected(self):
+        for change in ('duplicate', 'symlink', 'truncated'):
+            def bad_tree(url, etag=None):
+                body, tag = self.fetch(url, etag)
+                if '/git/trees/' in url:
+                    tree = json.loads(body)
+                    if change == 'duplicate': tree['tree'].append(tree['tree'][0])
+                    elif change == 'symlink': tree['tree'][0]['mode'] = '120000'
+                    else: tree['truncated'] = True
+                    body = json.dumps(tree).encode()
+                return body, tag
+            with self.subTest(change=change), self.assertRaises(ValueError):
+                refresh.synchronize(self.root, 'main', bad_tree)
+            self.assertEqual(refresh.current_catalog(self.root), self.root / 'catalog')
+
+    def test_bad_channel_config_falls_back_without_crashing_menu(self):
+        config = self.root.parent / 'catalog-source.json'
+        for value in ([], None, 'invalid', {'channel': 'other'}, {'beta_program': 'invalid'}):
+            config.write_text(json.dumps(value))
+            self.assertEqual(refresh.channel_for(self.root), 'main')
+
+    def test_full_disk_during_file_write_leaves_no_partial_catalog(self):
+        self.sync(); old = refresh.current_catalog(self.root)
+        self.revision = 'b' * 40; self.add_app()
+        real = Path.write_bytes
+        def full(path, data):
+            if any(part.startswith('.download-') for part in path.parts):
+                raise OSError(errno.ENOSPC, 'Disk full')
+            return real(path, data)
+        with patch.object(Path, 'write_bytes', new=full), self.assertRaises(OSError):
+            self.sync()
+        self.assertEqual(refresh.current_catalog(self.root), old)
+        self.assertFalse(list((refresh.cache_root(self.root) / 'snapshots').glob('.download-*')))
+
+    def test_orphaned_download_is_removed_before_retry(self):
+        partial = refresh.cache_root(self.root) / 'snapshots/.download-crashed'
+        partial.mkdir(parents=True); (partial/'partial.json').write_text('{}')
+        synchronize = refresh.synchronize
+        with patch.object(refresh, 'synchronize', side_effect=lambda *a, **kw:
+                          synchronize(self.root, 'main', self.fetch)):
+            refresh.worker(self.root, 'main')
+        self.assertFalse(partial.exists())
+        self.assertTrue(refresh.read_json(refresh.cache_root(self.root)/'main-result.json')['ok'])
+
+    def test_request_rejects_oversized_response_and_handles_304(self):
+        context = Mock()
+        context.__enter__ = Mock(return_value=Mock(read=Mock(return_value=b'oversized')))
+        context.__exit__ = Mock(return_value=False)
+        with patch.object(refresh.urllib.request, 'urlopen', return_value=context), self.assertRaises(ValueError):
+            refresh.request('https://api.github.com/test', limit=2)
+        error = urllib.error.HTTPError('https://api.github.com', 304, 'unchanged', {}, None)
+        with patch.object(refresh.urllib.request, 'urlopen', side_effect=error):
+            self.assertEqual(refresh.request('https://api.github.com/test', etag='old'), (None, 'old'))
+
+    def test_unwritable_seen_state_does_not_crash_menu(self):
+        status = refresh.CatalogRefresh(self.root, Catalog(self.root))
+        with patch.object(refresh, 'write_json', side_effect=PermissionError('Read only')):
+            status.mark_seen()
+        self.assertTrue(status.failed)
+
+    def test_concurrent_workers_serialize_the_catalog_transaction(self):
+        barrier = threading.Barrier(2)
+        guard = threading.Lock()
+        active = maximum = 0
+        errors = []
+        def operation(*args, **kwargs):
+            nonlocal active, maximum
+            with guard:
+                active += 1; maximum = max(maximum, active)
+            time.sleep(0.03)
+            with guard: active -= 1
+            return False
+        def run():
+            try:
+                barrier.wait(timeout=2)
+                refresh.worker(self.root, 'main')
+            except Exception as error:
+                errors.append(error)
+        with patch.object(refresh, 'synchronize', side_effect=operation):
+            threads = [threading.Thread(target=run) for _ in range(2)]
+            for thread in threads: thread.start()
+            for thread in threads: thread.join(timeout=3)
+        self.assertFalse(errors)
+        self.assertTrue(all(not thread.is_alive() for thread in threads))
+        self.assertEqual(maximum, 1)
+
+    def test_download_deadline_preserves_previous_catalog(self):
+        self.sync(); old = refresh.current_catalog(self.root)
+        self.revision = 'b' * 40; self.add_app()
+        with patch.object(refresh.time, 'monotonic', side_effect=itertools.chain([0], itertools.repeat(100))), \
+                self.assertRaises(TimeoutError):
+            self.sync()
+        self.assertEqual(refresh.current_catalog(self.root), old)
+
+    def test_http_end_to_end_new_release_outage_retry_manual_refresh_and_repair(self):
+        owner = self
+        unavailable = False
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def log_message(self, *args): pass
+            def do_GET(self):
+                if unavailable:
+                    self.send_error(503); return
+                etag = '"' + owner.revision + '"'
+                if self.headers.get('If-None-Match') == etag:
+                    self.send_response(304); self.end_headers(); return
+                host = ('https://api.github.com' if self.path.startswith('/repos/')
+                        else 'https://raw.githubusercontent.com')
+                body, _ = owner.fetch(host + self.path)
+                self.send_response(200)
+                self.send_header('Content-Length', str(len(body)))
+                self.send_header('ETag', etag)
+                self.end_headers(); self.wfile.write(body)
+        server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        def fetch(url, etag=None):
+            original = urllib.parse.urlsplit(url)
+            path = original.path + ('?' + original.query if original.query else '')
+            return refresh.request(f'http://127.0.0.1:{server.server_port}' + path, etag)
+        try:
+            self.assertTrue(refresh.synchronize(self.root, 'main', fetch))
+            self.assertFalse(refresh.synchronize(self.root, 'main', fetch))
+            first = refresh.current_catalog(self.root)
+            unavailable = True
+            self.revision = 'b' * 40; self.add_app()
+            with self.assertRaises(urllib.error.HTTPError):
+                refresh.synchronize(self.root, 'main', fetch)
+            self.assertEqual(refresh.current_catalog(self.root), first)
+            unavailable = False
+            self.assertTrue(refresh.synchronize(self.root, 'main', fetch))
+            current = Catalog(self.root)
+            self.assertEqual(len(refresh.visible_recipes(current)), 2)
+            self.assertFalse(refresh.synchronize(self.root, 'main', fetch, force=True))
+            (current.catalog_dir / 'curated/sample-added.json').write_text('{}')
+            self.assertTrue(refresh.synchronize(self.root, 'main', fetch, force=True))
+            self.assertEqual(Catalog(self.root).compose('sample-added')['id'], 'image-sample-added')
+        finally:
+            server.shutdown(); server.server_close(); thread.join(timeout=2)
+
 
 if __name__ == "__main__":
     unittest.main()