31 lines
1.1 KiB
Python
31 lines
1.1 KiB
Python
import json
|
|
|
|
import app.services.model_registry as registry_module
|
|
from app.services.model_registry import ModelRegistry
|
|
|
|
|
|
class FakeModel:
|
|
input_shape = (None, 7, 1)
|
|
|
|
|
|
def test_registry_skips_incomplete_artifacts(tmp_path, monkeypatch):
|
|
market_dir = tmp_path / "rawit" / "pasar_tanjung"
|
|
market_dir.mkdir(parents=True)
|
|
(market_dir / "config.json").write_text(
|
|
json.dumps({"model": {"time_step": 7}, "files": {"model": "model.h5", "scaler": "scaler.pkl"}}),
|
|
encoding="utf-8",
|
|
)
|
|
(market_dir / "model.h5").write_text("model", encoding="utf-8")
|
|
(market_dir / "scaler.pkl").write_bytes(b"scaler")
|
|
|
|
class FakeScaler:
|
|
data_min_ = [0]
|
|
data_max_ = [1]
|
|
|
|
monkeypatch.setattr(registry_module, "load_model", lambda *args, **kwargs: FakeModel())
|
|
monkeypatch.setattr(registry_module.joblib, "load", lambda *args, **kwargs: FakeScaler())
|
|
model_registry = ModelRegistry(tmp_path)
|
|
|
|
assert model_registry.get("rawit", "pasar_tanjung") is not None
|
|
assert any(item["market"] == "pasar_kreongan" for item in model_registry.skipped)
|