Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import os | |
| import shutil | |
| import subprocess | |
| import sys | |
| import unittest | |
| from pathlib import Path | |
| SERVICE_ROOT = Path(__file__).resolve().parents[1] | |
| REPO_ROOT = SERVICE_ROOT.parent.parent | |
| BACKEND_ROOT = SERVICE_ROOT.parent | |
| if str(SERVICE_ROOT) not in sys.path: | |
| sys.path.insert(0, str(SERVICE_ROOT)) | |
| from inference.config import manifest_sha256, reset_inference_manifest_cache # noqa: E402 | |
| CANONICAL_MANIFEST = BACKEND_ROOT / "config" / "inference.manifest.json" | |
| def _host_manifest_sha() -> str: | |
| return hashlib.sha256(CANONICAL_MANIFEST.read_bytes()).hexdigest() | |
| class ContainerManifestTests(unittest.TestCase): | |
| def setUp(self) -> None: | |
| reset_inference_manifest_cache() | |
| self._old_manifest_path = os.environ.pop("INFERENCE_MANIFEST_PATH", None) | |
| def tearDown(self) -> None: | |
| reset_inference_manifest_cache() | |
| if self._old_manifest_path is not None: | |
| os.environ["INFERENCE_MANIFEST_PATH"] = self._old_manifest_path | |
| else: | |
| os.environ.pop("INFERENCE_MANIFEST_PATH", None) | |
| def test_manifest_endpoint_contract(self) -> None: | |
| os.environ["INFERENCE_MANIFEST_PATH"] = str(CANONICAL_MANIFEST) | |
| self.assertEqual(manifest_sha256(), _host_manifest_sha()) | |
| def test_running_python_container_loads_same_manifest_sha(self) -> None: | |
| compose_file = REPO_ROOT / "docker-compose.yml" | |
| existing_container = subprocess.run( | |
| [ | |
| "docker", | |
| "compose", | |
| "-f", | |
| str(compose_file), | |
| "ps", | |
| "-q", | |
| "python-service", | |
| ], | |
| cwd=REPO_ROOT, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| was_running = False | |
| if existing_container: | |
| was_running = ( | |
| subprocess.run( | |
| [ | |
| "docker", | |
| "inspect", | |
| "-f", | |
| "{{.State.Running}}", | |
| existing_container, | |
| ], | |
| cwd=REPO_ROOT, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| == "true" | |
| ) | |
| config = subprocess.run( | |
| ["docker", "compose", "-f", str(compose_file), "config"], | |
| cwd=REPO_ROOT, | |
| capture_output=True, | |
| text=True, | |
| check=True, | |
| ) | |
| self.assertIn("INFERENCE_MANIFEST_PATH", config.stdout) | |
| self.assertIn("/config/inference.manifest.json", config.stdout) | |
| up = subprocess.run( | |
| [ | |
| "docker", | |
| "compose", | |
| "-f", | |
| str(compose_file), | |
| "up", | |
| "-d", | |
| "python-service", | |
| ], | |
| cwd=REPO_ROOT, | |
| capture_output=True, | |
| text=True, | |
| ) | |
| self.assertEqual(up.returncode, 0, up.stderr) | |
| try: | |
| exec_result = subprocess.run( | |
| [ | |
| "docker", | |
| "compose", | |
| "-f", | |
| str(compose_file), | |
| "exec", | |
| "-T", | |
| "python-service", | |
| "python", | |
| "-c", | |
| "from inference.config import manifest_sha256; print(manifest_sha256())", | |
| ], | |
| cwd=REPO_ROOT, | |
| capture_output=True, | |
| text=True, | |
| ) | |
| self.assertEqual(exec_result.returncode, 0, exec_result.stderr) | |
| container_sha = exec_result.stdout.strip() | |
| self.assertEqual(container_sha, _host_manifest_sha()) | |
| finally: | |
| if not was_running: | |
| subprocess.run( | |
| [ | |
| "docker", | |
| "compose", | |
| "-f", | |
| str(compose_file), | |
| "stop", | |
| "python-service", | |
| ], | |
| cwd=REPO_ROOT, | |
| capture_output=True, | |
| text=True, | |
| ) | |
| if __name__ == "__main__": | |
| unittest.main() | |