Spaces:
Running
Running
File size: 2,731 Bytes
a9a62dc 923beac a9a62dc 923beac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | from datetime import datetime, timezone
import inspect
from unittest.mock import patch
def test_worker_publishes_completed_ml_prediction():
from scripts.hf_queue_worker import _compute_stocks
now = datetime.now(timezone.utc)
ml_result = {
"signal": "BUY",
"source": "ml_model",
"signal_probability": 0.72,
}
with patch("services.predictor_service.get_ml_prediction_sync", return_value=ml_result):
queue, changed = _compute_stocks(["2330"], {}, now)
assert changed is True
assert queue["2330"]["status"] == "done"
assert queue["2330"]["source"] == "mac_worker"
assert queue["2330"]["result"] == ml_result
def test_worker_refuses_quick_rules_placeholder():
from scripts.hf_queue_worker import _compute_stocks
now = datetime.now(timezone.utc)
quick_result = {
"signal": "HOLD",
"source": "quick_rules",
"signal_probability": 0.5,
}
with patch("services.predictor_service.get_ml_prediction_sync", return_value=quick_result):
queue, changed = _compute_stocks(["2330"], {}, now)
assert changed is True
assert queue["2330"]["status"] == "error"
assert queue["2330"]["source"] == "mac_worker"
assert "quick_rules" in queue["2330"]["error"]
def test_queue_cache_rejects_quick_rules_result():
from services.hf_queue_service import get_cached_result, is_publishable_result
queue = {
"2330": {
"status": "done",
"completed_at": datetime.now(timezone.utc).isoformat(),
"result": {
"signal": "HOLD",
"source": "quick_rules",
"disclaimer": "Quick estimate (technical rules only).",
},
}
}
with patch("services.hf_queue_service.read_queue", return_value=queue):
assert is_publishable_result(queue["2330"]["result"]) is False
assert get_cached_result("2330") is None
def test_queue_cache_accepts_ml_result():
from services.hf_queue_service import get_cached_result, is_publishable_result
result = {
"signal": "BUY",
"source": "ml_model",
"signal_probability": 0.72,
}
queue = {
"2330": {
"status": "done",
"completed_at": datetime.now(timezone.utc).isoformat(),
"result": result,
}
}
with patch("services.hf_queue_service.read_queue", return_value=queue):
assert is_publishable_result(result) is True
assert get_cached_result("2330") == result
def test_hf_startup_warmup_filters_placeholders():
from main import _warmup_from_queue
source = inspect.getsource(_warmup_from_queue)
assert "is_publishable_result" in source
|