Compare commits
12 Commits
2.0.3
...
29700169ad
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
29700169ad | ||
|
|
dfbdd4b605 | ||
|
|
0f5e4a5cef | ||
|
|
993c7d7662 | ||
| dc9fd871bb | |||
| 242566f9b1 | |||
|
|
7903dda730 | ||
| d0b4644f63 | |||
| ced847bbe7 | |||
|
|
262ba8d658 | ||
|
|
0c9c5f300b | ||
| 49c8942a87 |
15
.github/workflows/deploy.yml
vendored
15
.github/workflows/deploy.yml
vendored
@@ -1,10 +1,15 @@
|
|||||||
name: Deploy DataOps Worker
|
name: Deploy DataOps Worker
|
||||||
|
|
||||||
on:
|
on:
|
||||||
pull_request:
|
|
||||||
types: [closed]
|
|
||||||
branches: [main, 'release/**', 'feature/**']
|
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
inputs:
|
||||||
|
chart_ref:
|
||||||
|
description: >
|
||||||
|
sientia-helm-chart branch to bump GITHUB_BRANCH on AND deploy — the literal
|
||||||
|
deploy target. Validated via git ls-remote; a branch that does not exist in
|
||||||
|
sientia-helm-chart is a hard error, not a silent fallback to main.
|
||||||
|
required: true
|
||||||
|
type: string
|
||||||
|
|
||||||
permissions:
|
permissions:
|
||||||
contents: write
|
contents: write
|
||||||
@@ -14,10 +19,10 @@ permissions:
|
|||||||
jobs:
|
jobs:
|
||||||
deploy:
|
deploy:
|
||||||
name: Deploy
|
name: Deploy
|
||||||
if: github.event_name == 'workflow_dispatch' || github.event.pull_request.merged == true
|
|
||||||
uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
|
uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
|
||||||
with:
|
with:
|
||||||
branch_name: ${{ github.event.pull_request.head.ref }}
|
branch_name: ${{ github.ref_name }}
|
||||||
|
chart_ref: ${{ inputs.chart_ref }}
|
||||||
project_name: 'model-manager'
|
project_name: 'model-manager'
|
||||||
chart_name: 'model-manager'
|
chart_name: 'model-manager'
|
||||||
env_list_path: '[".global.env"]'
|
env_list_path: '[".global.env"]'
|
||||||
|
|||||||
1
.github/workflows/quality-gate.yml
vendored
1
.github/workflows/quality-gate.yml
vendored
@@ -5,7 +5,6 @@ on:
|
|||||||
branches:
|
branches:
|
||||||
- main
|
- main
|
||||||
- 'release/**'
|
- 'release/**'
|
||||||
- 'feature/**'
|
|
||||||
pull_request:
|
pull_request:
|
||||||
branches:
|
branches:
|
||||||
- main
|
- main
|
||||||
|
|||||||
16
.github/workflows/require-chart-ref-label.yml
vendored
16
.github/workflows/require-chart-ref-label.yml
vendored
@@ -1,16 +0,0 @@
|
|||||||
name: Require chart-ref label
|
|
||||||
|
|
||||||
on:
|
|
||||||
pull_request:
|
|
||||||
types: [opened, synchronize, reopened, labeled, unlabeled]
|
|
||||||
branches:
|
|
||||||
- main
|
|
||||||
- 'release/**'
|
|
||||||
- 'feature/**'
|
|
||||||
|
|
||||||
permissions:
|
|
||||||
pull-requests: write
|
|
||||||
|
|
||||||
jobs:
|
|
||||||
require-chart-ref-label:
|
|
||||||
uses: Aignosi/github_workflow_templates/.github/workflows/require-chart-ref-label.yml@main
|
|
||||||
17
.github/workflows/version-bump.yml
vendored
Normal file
17
.github/workflows/version-bump.yml
vendored
Normal file
@@ -0,0 +1,17 @@
|
|||||||
|
name: Version Bump
|
||||||
|
|
||||||
|
permissions:
|
||||||
|
contents: write
|
||||||
|
|
||||||
|
on:
|
||||||
|
pull_request:
|
||||||
|
types: [closed]
|
||||||
|
branches: [main, 'release/**', 'feature/**']
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
bump:
|
||||||
|
if: github.event.pull_request.merged == true
|
||||||
|
uses: Aignosi/github_workflow_templates/.github/workflows/reusable-version-bump.yml@main
|
||||||
|
with:
|
||||||
|
update_version_in: '["pyproject"]'
|
||||||
|
secrets: inherit
|
||||||
@@ -45,7 +45,7 @@ Coverage gate in CI is 80% minimum; this repo targets 100% in practice — don't
|
|||||||
## Architecture
|
## Architecture
|
||||||
|
|
||||||
### Temporal layering
|
### Temporal layering
|
||||||
- **Worker** (`model_manager/worker/worker.py`): entrypoint. Builds config from env, connects Postgres/MLFlow/MinIO/Plugin Store, installs the `RUNTIME` model runtime via `PluginStore.install_runtime`, starts two Temporal workers (one per task queue) plus Prometheus metrics server and the cleanup schedule. All heavy imports are wrapped in `workflow.unsafe.imports_passed_through()` — required because Temporal workflow sandboxing forbids non-deterministic imports at module scope outside that block.
|
- **Worker** (`model_manager/worker/worker.py`): entrypoint. Builds config from env, connects Postgres/MLFlow/MinIO/Plugin Store, installs the `RUNTIME` model runtime via `PluginStore.install_runtime` (boot fails if that changes the version of an already-imported package, so the `requirements.txt` pins must match the factory runtimes; `app_up` goes to 1 only after this check), starts two Temporal workers (one per task queue) plus Prometheus metrics server and the cleanup schedule. All heavy imports are wrapped in `workflow.unsafe.imports_passed_through()` — required because Temporal workflow sandboxing forbids non-deterministic imports at module scope outside that block.
|
||||||
- **Workflows** (`model_manager/workflows/`): pure orchestration, no I/O beyond `execute_activity_method` calls.
|
- **Workflows** (`model_manager/workflows/`): pure orchestration, no I/O beyond `execute_activity_method` calls.
|
||||||
- `train_model.py` (`TrainModel`, workflow name `train_model`): validate `experiment_run_id` → `load_model_metadata` → `validate_train_params` → update DB status `ORCHESTRATOR_WAITING_PROC` → `train_model` activity → update DB status `TRAINING_SUCCESS`/`TRAINING_ERROR` → `cleanup_resources` (always runs in `finally` if a `run_dir` exists, even on failure). Uses distinct `RetryPolicy`s per operation class: `network_retry_policy` (transient I/O), `no_retry_policy` (training/validation — data errors are permanent, and retrying would duplicate MLflow writes), `database_retry_policy` (Postgres). Timeouts are env-driven (`TIMEOUT_*`), extend `TIMEOUT_TRAIN_MODEL` instead of adding retries for slow training.
|
- `train_model.py` (`TrainModel`, workflow name `train_model`): validate `experiment_run_id` → `load_model_metadata` → `validate_train_params` → update DB status `ORCHESTRATOR_WAITING_PROC` → `train_model` activity → update DB status `TRAINING_SUCCESS`/`TRAINING_ERROR` → `cleanup_resources` (always runs in `finally` if a `run_dir` exists, even on failure). Uses distinct `RetryPolicy`s per operation class: `network_retry_policy` (transient I/O), `no_retry_policy` (training/validation — data errors are permanent, and retrying would duplicate MLflow writes), `database_retry_policy` (Postgres). Timeouts are env-driven (`TIMEOUT_*`), extend `TIMEOUT_TRAIN_MODEL` instead of adding retries for slow training.
|
||||||
- `cleanup_files.py` (`CleanupFiles`): scheduled (`cleanup_schedule.py`, default daily at midnight UTC via `CLEANUP_CRON`) deletion of stale local temp dirs matching `name_YYYYMMDD_HHMMSS_microseconds`, older than `CLEANUP_RETENTION_HOURS`.
|
- `cleanup_files.py` (`CleanupFiles`): scheduled (`cleanup_schedule.py`, default daily at midnight UTC via `CLEANUP_CRON`) deletion of stale local temp dirs matching `name_YYYYMMDD_HHMMSS_microseconds`, older than `CLEANUP_RETENTION_HOURS`.
|
||||||
|
|||||||
@@ -149,6 +149,7 @@ The Model Manager system uses a Temporal-based workflow architecture with clear
|
|||||||
- Prometheus metrics server initialization
|
- Prometheus metrics server initialization
|
||||||
- Notification handler setup and configuration
|
- Notification handler setup and configuration
|
||||||
- **Runtime Installation**: Automatically installs the required model runtime from the Plugin Store
|
- **Runtime Installation**: Automatically installs the required model runtime from the Plugin Store
|
||||||
|
- **Runtime version guard**: boot fails if the install changes the version of a package the worker already imported, so the `requirements.txt` pins must match `model-plugin-factory` `runtime/basic.yaml` and `runtime/xgboost.yaml`; `app_up` goes to 1 only after this check
|
||||||
- **Key Features**:
|
- **Key Features**:
|
||||||
- Automatic scaling with `PollerBehaviorAutoscaling`
|
- Automatic scaling with `PollerBehaviorAutoscaling`
|
||||||
- Health check endpoints for Kubernetes liveness/readiness probes
|
- Health check endpoints for Kubernetes liveness/readiness probes
|
||||||
@@ -813,7 +814,9 @@ Runs automatically on each Pull Request to `main`:
|
|||||||
- ✅ Automatic version calculation based on branch pattern
|
- ✅ Automatic version calculation based on branch pattern
|
||||||
|
|
||||||
#### Deploy (`.github/workflows/deploy.yml`)
|
#### Deploy (`.github/workflows/deploy.yml`)
|
||||||
Runs automatically when a PR is merged to `main`:
|
Triggered manually via `workflow_dispatch`, with a required `chart_ref` input (no
|
||||||
|
default — the sientia-helm-chart branch to bump and deploy; a branch that doesn't
|
||||||
|
exist is a hard error, not a silent fallback):
|
||||||
- ✅ Builds and pushes Docker image to Azure Container Registry
|
- ✅ Builds and pushes Docker image to Azure Container Registry
|
||||||
- ✅ Creates GitHub release with calculated version
|
- ✅ Creates GitHub release with calculated version
|
||||||
- ✅ Deploys to Kubernetes using Helm
|
- ✅ Deploys to Kubernetes using Helm
|
||||||
@@ -1247,6 +1250,10 @@ sientia-dataops-model-manager/
|
|||||||
- Check training parameter validation errors
|
- Check training parameter validation errors
|
||||||
- Verify input data format and required fields
|
- Verify input data format and required fields
|
||||||
|
|
||||||
|
5. **Worker exits at boot with `Runtime ... changed packages already imported by the worker`**
|
||||||
|
- A runtime pin in `model-plugin-factory` `runtime/*.yaml` no longer matches `requirements.txt`; the message lists each package as `old -> new`
|
||||||
|
- Set the runtime version in `requirements.txt` and `requirements-local.txt` here and in `sientia-dataops-laborious`, then redeploy
|
||||||
|
|
||||||
### Debug Mode
|
### Debug Mode
|
||||||
|
|
||||||
Enable debug logging by setting the log level:
|
Enable debug logging by setting the log level:
|
||||||
|
|||||||
@@ -32,6 +32,7 @@ from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
|
|||||||
|
|
||||||
with workflow.unsafe.imports_passed_through():
|
with workflow.unsafe.imports_passed_through():
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import importlib.metadata
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
@@ -76,6 +77,36 @@ def _get_runtime(runtime: str | None) -> str:
|
|||||||
return normalized_runtime or 'single'
|
return normalized_runtime or 'single'
|
||||||
|
|
||||||
|
|
||||||
|
def _loaded_distribution_versions() -> dict[str, str]:
|
||||||
|
"""Version on disk of every installed distribution whose top-level module is imported."""
|
||||||
|
# Iterate a copy: another thread importing while we iterate would raise RuntimeError.
|
||||||
|
loaded = {name.partition('.')[0] for name in sys.modules.copy()}
|
||||||
|
return {
|
||||||
|
dist: importlib.metadata.version(dist)
|
||||||
|
for module, dists in importlib.metadata.packages_distributions().items()
|
||||||
|
if module in loaded
|
||||||
|
for dist in dists
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _runtime_changed_loaded_packages(before: dict[str, str]) -> dict[str, str]:
|
||||||
|
"""
|
||||||
|
Packages that were imported before ``install_runtime`` and whose version changed on disk.
|
||||||
|
|
||||||
|
The process keeps the old code in memory, so a later lazy import mixes two versions
|
||||||
|
(SIENTIAPDE-2214: sklearn 1.9.1 in memory, 1.8.0 on disk, ``_cyutility`` import error).
|
||||||
|
"""
|
||||||
|
changed = {}
|
||||||
|
for dist, old in before.items():
|
||||||
|
try:
|
||||||
|
new = importlib.metadata.version(dist)
|
||||||
|
except importlib.metadata.PackageNotFoundError:
|
||||||
|
new = 'removed'
|
||||||
|
if new != old:
|
||||||
|
changed[dist] = f'{old} -> {new}'
|
||||||
|
return changed
|
||||||
|
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
"""
|
"""
|
||||||
Main entry point for the Model Manager worker application.
|
Main entry point for the Model Manager worker application.
|
||||||
@@ -83,7 +114,9 @@ async def main():
|
|||||||
This function initializes and starts all components of the worker:
|
This function initializes and starts all components of the worker:
|
||||||
1. Sets up logging and metadata
|
1. Sets up logging and metadata
|
||||||
2. Starts Prometheus metrics server
|
2. Starts Prometheus metrics server
|
||||||
3. Initializes notification handler
|
3. Initializes notification handler, then installs the runtime; exits if the install
|
||||||
|
changed the version of a package the worker already imported, and only then sets
|
||||||
|
``app_up`` (what the probes check) to 1
|
||||||
4. Creates and configures activities
|
4. Creates and configures activities
|
||||||
5. Starts Temporal client and workers
|
5. Starts Temporal client and workers
|
||||||
6. Manages worker lifecycle and graceful shutdown
|
6. Manages worker lifecycle and graceful shutdown
|
||||||
@@ -144,8 +177,23 @@ async def main():
|
|||||||
metrics_controller=metrics_controller,
|
metrics_controller=metrics_controller,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
loaded_before_install = _loaded_distribution_versions()
|
||||||
await plugin_store.install_runtime(runtime_name=runtime)
|
await plugin_store.install_runtime(runtime_name=runtime)
|
||||||
|
|
||||||
|
changed = _runtime_changed_loaded_packages(loaded_before_install)
|
||||||
|
if changed:
|
||||||
|
logger.custom_critical(
|
||||||
|
f'Runtime {runtime} changed packages already imported by the worker: {changed}. '
|
||||||
|
'Pin them in requirements.txt to the runtime versions.',
|
||||||
|
metadata,
|
||||||
|
)
|
||||||
|
metrics.APP_UP.labels(pod_id=POD_ID).set(0)
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
# The probes only check app_up, so it goes to 1 after the guard: a pod that would fail it
|
||||||
|
# never turns Ready, and a rollout keeps the old pod.
|
||||||
|
metrics.APP_UP.labels(pod_id=POD_ID).set(1)
|
||||||
|
|
||||||
activities = Activities(
|
activities = Activities(
|
||||||
postgres_config=build_postgres_config(),
|
postgres_config=build_postgres_config(),
|
||||||
mlflow_config=build_mlflow_config(),
|
mlflow_config=build_mlflow_config(),
|
||||||
@@ -232,8 +280,8 @@ def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | Non
|
|||||||
"""
|
"""
|
||||||
Starts the Prometheus metrics server for monitoring and observability.
|
Starts the Prometheus metrics server for monitoring and observability.
|
||||||
|
|
||||||
This function initializes the Prometheus HTTP server on the configured port
|
This function initializes the Prometheus HTTP server on the configured port.
|
||||||
and sets the application health metric to indicate the service is running.
|
``main`` sets ``app_up`` to 1 later, once the runtime install passed the version guard.
|
||||||
|
|
||||||
The server exposes metrics that can be scraped by Prometheus for monitoring
|
The server exposes metrics that can be scraped by Prometheus for monitoring
|
||||||
the health and performance of the Model Manager worker.
|
the health and performance of the Model Manager worker.
|
||||||
@@ -249,7 +297,6 @@ def start_prometheus_server(logger: SientiaLogger, metadata: dict[str, str | Non
|
|||||||
port = int(os.getenv('HTTP_METRICS_PORT', 9090))
|
port = int(os.getenv('HTTP_METRICS_PORT', 9090))
|
||||||
start_http_server(port)
|
start_http_server(port)
|
||||||
logger.custom_info(f'Prometheus server initialized on port {port}.', metadata)
|
logger.custom_info(f'Prometheus server initialized on port {port}.', metadata)
|
||||||
metrics.APP_UP.labels(pod_id=POD_ID).set(1) # Mark app as UP
|
|
||||||
except Exception as e: # noqa: BLE001
|
except Exception as e: # noqa: BLE001
|
||||||
logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata)
|
logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata)
|
||||||
os._exit(1)
|
os._exit(1)
|
||||||
|
|||||||
@@ -8,4 +8,15 @@ git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.14.0
|
|||||||
prometheus-client==0.23.1
|
prometheus-client==0.23.1
|
||||||
beautifulsoup4==4.12.3
|
beautifulsoup4==4.12.3
|
||||||
evidently==0.6.7
|
evidently==0.6.7
|
||||||
jsonschema==4.26.0
|
jsonschema==4.26.0
|
||||||
|
# Same versions as the plugin runtimes (model-plugin-factory runtime/basic.yaml and xgboost.yaml).
|
||||||
|
# The worker imports these before install_runtime; if the runtime installed a different version,
|
||||||
|
# the old code would stay in memory, so the worker refuses to boot (SIENTIAPDE-2214).
|
||||||
|
# Changing a runtime pin means changing it here and in requirements.txt too.
|
||||||
|
mlflow==3.8.1
|
||||||
|
pathspec==1.0.4
|
||||||
|
pandas==2.3.3
|
||||||
|
cloudpickle==3.1.2
|
||||||
|
scikit-learn==1.8.0
|
||||||
|
joblib==1.5.3
|
||||||
|
rich==14.3.2
|
||||||
|
|||||||
@@ -4,8 +4,19 @@ sqlalchemy==2.0.50
|
|||||||
boto3==1.42.70
|
boto3==1.42.70
|
||||||
botocore==1.42.70
|
botocore==1.42.70
|
||||||
sientia_do>=1.12.2
|
sientia_do>=1.12.2
|
||||||
sientia_model>=0.8.1
|
sientia_model>=0.14.0
|
||||||
prometheus-client==0.23.1
|
prometheus-client==0.23.1
|
||||||
beautifulsoup4==4.12.3
|
beautifulsoup4==4.12.3
|
||||||
evidently==0.6.7
|
evidently==0.6.7
|
||||||
jsonschema==4.26.0
|
jsonschema==4.26.0
|
||||||
|
# Same versions as the plugin runtimes (model-plugin-factory runtime/basic.yaml and xgboost.yaml).
|
||||||
|
# The worker imports these before install_runtime; if the runtime installed a different version,
|
||||||
|
# the old code would stay in memory, so the worker refuses to boot (SIENTIAPDE-2214).
|
||||||
|
# Changing a runtime pin means changing it here and in requirements-local.txt too.
|
||||||
|
mlflow==3.8.1
|
||||||
|
pathspec==1.0.4
|
||||||
|
pandas==2.3.3
|
||||||
|
cloudpickle==3.1.2
|
||||||
|
scikit-learn==1.8.0
|
||||||
|
joblib==1.5.3
|
||||||
|
rich==14.3.2
|
||||||
|
|||||||
@@ -3,7 +3,7 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
from unittest.mock import AsyncMock, Mock, patch
|
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
@@ -132,9 +132,8 @@ def test_start_prometheus_server_success(
|
|||||||
# Verify HTTP server started
|
# Verify HTTP server started
|
||||||
mock_start_http_server.assert_called_once_with(9090)
|
mock_start_http_server.assert_called_once_with(9090)
|
||||||
|
|
||||||
# Verify APP_UP metric was set to 1
|
# app_up goes to 1 in main(), after the runtime version guard.
|
||||||
mock_metrics.APP_UP.labels.assert_called_once_with(pod_id='test-pod-123')
|
mock_app_up.set.assert_not_called()
|
||||||
mock_app_up.set.assert_called_once_with(1)
|
|
||||||
mock_logger.custom_info.assert_called_once()
|
mock_logger.custom_info.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
@@ -284,7 +283,8 @@ async def test_main_successful_startup(
|
|||||||
# Verify cleanup was performed
|
# Verify cleanup was performed
|
||||||
mock_notification_handler.shutdown.assert_called_once()
|
mock_notification_handler.shutdown.assert_called_once()
|
||||||
mock_activities.shutdown.assert_called_once()
|
mock_activities.shutdown.assert_called_once()
|
||||||
mock_app_up.set.assert_called_with(0)
|
# Ready after the guard, down again on shutdown.
|
||||||
|
assert [c.args[0] for c in mock_app_up.set.call_args_list] == [1, 0]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
@@ -848,12 +848,10 @@ async def test_main_missing_runtime_uses_single_fallback(
|
|||||||
def test_main_entrypoint(mock_asyncio_run):
|
def test_main_entrypoint(mock_asyncio_run):
|
||||||
"""Test the __main__ entrypoint."""
|
"""Test the __main__ entrypoint."""
|
||||||
# Import and execute the main block
|
# Import and execute the main block
|
||||||
with patch.object(sys, 'argv', ['worker.py']):
|
import model_manager.worker.worker as worker_module
|
||||||
import model_manager.worker.worker as worker_module
|
|
||||||
|
|
||||||
# Simulate running the module
|
|
||||||
worker_module.main = AsyncMock()
|
|
||||||
|
|
||||||
|
# patch.object restores main afterwards; assigning it leaked the mock into later tests.
|
||||||
|
with patch.object(sys, 'argv', ['worker.py']), patch.object(worker_module, 'main', AsyncMock()):
|
||||||
# This would normally be called by asyncio.run(main())
|
# This would normally be called by asyncio.run(main())
|
||||||
# We just verify the pattern is correct
|
# We just verify the pattern is correct
|
||||||
assert callable(worker_module.main)
|
assert callable(worker_module.main)
|
||||||
@@ -905,3 +903,105 @@ def test_start_prometheus_server_prints_failure(
|
|||||||
# Agora o erro é logado via logger crítico
|
# Agora o erro é logado via logger crítico
|
||||||
mock_logger.custom_critical.assert_called_once()
|
mock_logger.custom_critical.assert_called_once()
|
||||||
mock_exit.assert_called_once_with(1)
|
mock_exit.assert_called_once_with(1)
|
||||||
|
|
||||||
|
|
||||||
|
def test_loaded_distribution_versions_reads_imported_packages():
|
||||||
|
"""An imported module maps to its distribution's installed version."""
|
||||||
|
import importlib.metadata
|
||||||
|
|
||||||
|
from model_manager.worker.worker import _loaded_distribution_versions
|
||||||
|
|
||||||
|
assert _loaded_distribution_versions()['pytest'] == importlib.metadata.version('pytest')
|
||||||
|
|
||||||
|
|
||||||
|
def test_runtime_changed_loaded_packages_reports_only_changes():
|
||||||
|
"""Only distributions whose version on disk differs after the install are reported."""
|
||||||
|
import importlib.metadata
|
||||||
|
|
||||||
|
from model_manager.worker.worker import _runtime_changed_loaded_packages
|
||||||
|
|
||||||
|
on_disk = {'scikit-learn': '1.8.0', 'numpy': '1.26.4'}
|
||||||
|
|
||||||
|
def fake_version(dist):
|
||||||
|
if dist not in on_disk:
|
||||||
|
raise importlib.metadata.PackageNotFoundError(dist)
|
||||||
|
return on_disk[dist]
|
||||||
|
|
||||||
|
before = {'scikit-learn': '1.9.1', 'numpy': '1.26.4', 'gone': '1.0'}
|
||||||
|
with patch('model_manager.worker.worker.importlib.metadata.version', side_effect=fake_version):
|
||||||
|
changed = _runtime_changed_loaded_packages(before)
|
||||||
|
|
||||||
|
assert changed == {'scikit-learn': '1.9.1 -> 1.8.0', 'gone': '1.0 -> removed'}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_main_runtime_changed_loaded_package_exits(mock_env_vars, mock_logger):
|
||||||
|
"""Boot stops before Activities when install_runtime changed an imported package."""
|
||||||
|
import model_manager.worker.worker as worker_module
|
||||||
|
|
||||||
|
plugin_store = Mock()
|
||||||
|
plugin_store.install_runtime = AsyncMock(return_value=None)
|
||||||
|
before = {'scikit-learn': '1.9.1'}
|
||||||
|
|
||||||
|
def snapshot():
|
||||||
|
# Taken after install_runtime, it would already see 1.8.0 and the guard would pass.
|
||||||
|
assert plugin_store.install_runtime.await_count == 0
|
||||||
|
return before
|
||||||
|
|
||||||
|
mongo = {'connection_string': 'mongodb://test', 'database_name': 'test_db', 'uri': 'x'}
|
||||||
|
with (
|
||||||
|
patch.object(worker_module, 'RUNTIME', 'basic'),
|
||||||
|
patch.object(worker_module, 'ensure_runtime_directories'),
|
||||||
|
patch.object(worker_module, 'metrics') as mock_metrics,
|
||||||
|
patch.object(worker_module, 'start_prometheus_server'),
|
||||||
|
patch.object(worker_module, 'get_logger', return_value=mock_logger),
|
||||||
|
patch.object(worker_module, 'build_mongodb_config', return_value=mongo),
|
||||||
|
patch.object(worker_module, 'NotificationHandler'),
|
||||||
|
patch.object(worker_module, 'build_plugin_store_config', return_value=MagicMock()),
|
||||||
|
patch.object(worker_module, 'PluginStore', return_value=plugin_store),
|
||||||
|
patch.object(worker_module, '_loaded_distribution_versions', side_effect=snapshot),
|
||||||
|
patch.object(
|
||||||
|
worker_module,
|
||||||
|
'_runtime_changed_loaded_packages',
|
||||||
|
return_value={'scikit-learn': '1.9.1 -> 1.8.0'},
|
||||||
|
) as mock_changed,
|
||||||
|
patch.object(worker_module, 'Activities') as mock_activities_class,
|
||||||
|
patch.object(worker_module.sys, 'exit', side_effect=SystemExit(1)) as mock_exit,
|
||||||
|
):
|
||||||
|
with pytest.raises(SystemExit):
|
||||||
|
await worker_module.main()
|
||||||
|
|
||||||
|
mock_exit.assert_called_once_with(1)
|
||||||
|
mock_activities_class.assert_not_called()
|
||||||
|
plugin_store.install_runtime.assert_awaited_once()
|
||||||
|
mock_changed.assert_called_once_with(before)
|
||||||
|
mock_metrics.APP_UP.labels.return_value.set.assert_called_once_with(0)
|
||||||
|
message = mock_logger.custom_critical.call_args.args[0]
|
||||||
|
assert "'scikit-learn': '1.9.1 -> 1.8.0'" in message
|
||||||
|
|
||||||
|
|
||||||
|
def test_guard_detects_version_change_on_disk(tmp_path, monkeypatch):
|
||||||
|
"""The real importlib.metadata path: a package swapped on disk after import is reported."""
|
||||||
|
import types
|
||||||
|
|
||||||
|
from model_manager.worker.worker import (
|
||||||
|
_loaded_distribution_versions,
|
||||||
|
_runtime_changed_loaded_packages,
|
||||||
|
)
|
||||||
|
|
||||||
|
info = tmp_path / 'fakepkg_2214-1.0.dist-info'
|
||||||
|
info.mkdir()
|
||||||
|
(info / 'METADATA').write_text('Metadata-Version: 2.1\nName: fakepkg-2214\nVersion: 1.0\n')
|
||||||
|
(info / 'top_level.txt').write_text('fakepkg_2214\n')
|
||||||
|
monkeypatch.syspath_prepend(str(tmp_path))
|
||||||
|
# The guard only reads sys.modules keys; setitem removes the entry again on teardown.
|
||||||
|
monkeypatch.setitem(sys.modules, 'fakepkg_2214', types.ModuleType('fakepkg_2214'))
|
||||||
|
|
||||||
|
before = _loaded_distribution_versions()
|
||||||
|
assert before['fakepkg-2214'] == '1.0'
|
||||||
|
|
||||||
|
# What pip does on an upgrade: the old dist-info goes away and a new one takes its place.
|
||||||
|
new_info = info.rename(tmp_path / 'fakepkg_2214-2.0.dist-info')
|
||||||
|
(new_info / 'METADATA').write_text('Metadata-Version: 2.1\nName: fakepkg-2214\nVersion: 2.0\n')
|
||||||
|
|
||||||
|
assert _runtime_changed_loaded_packages(before) == {'fakepkg-2214': '1.0 -> 2.0'}
|
||||||
|
|||||||
Reference in New Issue
Block a user