203 lines
7.4 KiB
Python
203 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
"""SMC model autotest — проверяет все загруженные модели классификации после обновления.
|
||
|
||
Использование:
|
||
python3 test_smc_models.py
|
||
python3 test_smc_models.py --host 10.101.60.27
|
||
python3 test_smc_models.py --host 10.101.60.27 --port 6181
|
||
python3 test_smc_models.py --asserts smc_asserts.json
|
||
|
||
Файл ассертов (JSON):
|
||
{
|
||
"asserts": [
|
||
{
|
||
"model": "МФЦ НСК", // подстрока для поиска модели
|
||
"type": "classifier",
|
||
"text": "записаться на приём", // опционально, свой текст
|
||
"checks": {
|
||
"min_classes": 1, // мин. количество классов
|
||
"top_class_contains": "МФЦ", // подстрока в top классе
|
||
"confidence_min": 0.5 // мин. confidence топ-класса
|
||
}
|
||
}
|
||
]
|
||
}
|
||
|
||
Возвращает exit code 0 если все модели прошли, иначе 1.
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import unittest
|
||
from urllib.parse import quote
|
||
|
||
try:
|
||
import urllib.request as urlreq
|
||
import urllib.error as urlerr
|
||
except ImportError:
|
||
urlreq = None
|
||
|
||
DEFAULT_HOST = "10.101.60.27"
|
||
DEFAULT_PORT = 6181
|
||
BASE_PATH = "/smc"
|
||
|
||
DEFAULT_TEXT = "Здравствуйте, хочу записаться на приём в МФЦ для получения паспорта"
|
||
|
||
|
||
def load_asserts(path):
|
||
if not path:
|
||
return []
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return data.get("asserts", [])
|
||
|
||
|
||
class SmcAPI:
|
||
def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT):
|
||
self.base = f"http://{host}:{port}{BASE_PATH}"
|
||
|
||
def get_models(self):
|
||
return json.loads(self._request(f"{self.base}/models"))
|
||
|
||
def classify(self, model_id, text):
|
||
encoded = quote(model_id, safe="")
|
||
url = f"{self.base}/classify/{encoded}?text={quote(text)}"
|
||
return json.loads(self._request(url))
|
||
|
||
def _request(self, url, timeout=30):
|
||
try:
|
||
with urlreq.urlopen(url, timeout=timeout) as r:
|
||
return r.read().decode("utf-8")
|
||
except urlerr.HTTPError as e:
|
||
raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8')[:200]}")
|
||
except urlerr.URLError as e:
|
||
raise RuntimeError(f"Connection failed: {e.reason}")
|
||
|
||
|
||
class TestSmcModels(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.api = SmcAPI(cls.args.host, cls.args.port)
|
||
cls.asserts = load_asserts(cls.args.asserts)
|
||
data = cls.api.get_models()
|
||
TestSmcModels.models = data
|
||
models = data.get("models", [])
|
||
handlers = data.get("handlers", [])
|
||
print(f"\n models: {len(models)}, handlers: {len(handlers)}")
|
||
if cls.asserts:
|
||
print(f" asserts loaded: {len(cls.asserts)}")
|
||
if len(models) + len(handlers) == 0:
|
||
raise unittest.SkipTest("Нет ни одной модели")
|
||
|
||
def _find_assert(self, model_id):
|
||
for a in self.asserts:
|
||
if a["model"] in model_id:
|
||
return a
|
||
return None
|
||
|
||
def _check_assert(self, model_id, result):
|
||
a = self._find_assert(model_id)
|
||
if not a:
|
||
return
|
||
checks = a.get("checks", {})
|
||
classes = result.get("classes", [])
|
||
if "min_found" in checks:
|
||
self.assertGreaterEqual(len(classes), checks["min_found"],
|
||
f"{model_id}: expected >= {checks['min_found']} classes, got {len(classes)}")
|
||
if classes and "confidence_min" in checks:
|
||
self.assertGreaterEqual(classes[0].get("confidence", 0), checks["confidence_min"],
|
||
f"{model_id}: top confidence < {checks['confidence_min']}")
|
||
|
||
def test_01_list_models(self):
|
||
"""Получение списка моделей"""
|
||
data = self.models
|
||
self.assertIn("models", data)
|
||
self.assertIn("handlers", data)
|
||
print(f"\n models: {len(data['models'])}")
|
||
print(f" handlers: {len(data['handlers'])}")
|
||
|
||
def test_02_models(self):
|
||
"""Классификация через модели"""
|
||
errors = []
|
||
models = self.models.get("models", [])
|
||
for mid in models:
|
||
if not mid:
|
||
continue
|
||
text = DEFAULT_TEXT
|
||
a = self._find_assert(mid)
|
||
if a and "text" in a:
|
||
text = a["text"]
|
||
try:
|
||
result = self.api.classify(mid, text)
|
||
classes = result.get("classes", [])
|
||
if not classes:
|
||
print(f" [!] {mid}: пустой результат")
|
||
else:
|
||
self._check_assert(mid, result)
|
||
print(f" >> {mid}")
|
||
for c in classes[:5]:
|
||
cl = c.get("class", "?")
|
||
conf = c.get("confidence", 0)
|
||
print(f" class='{cl}' conf={conf:.3f}")
|
||
if len(classes) > 5:
|
||
print(f" ... ({len(classes)} classes)")
|
||
except AssertionError:
|
||
raise
|
||
except Exception as e:
|
||
errors.append(f" {mid}: {e}")
|
||
if errors:
|
||
self.fail("Ошибки моделей:\n" + "\n".join(errors))
|
||
|
||
def test_03_handlers(self):
|
||
"""Классификация через handler модели"""
|
||
errors = []
|
||
handlers = self.models.get("handlers", [])
|
||
if not handlers:
|
||
self.skipTest("Нет handler моделей")
|
||
for mid in handlers:
|
||
if not mid:
|
||
continue
|
||
text = DEFAULT_TEXT
|
||
a = self._find_assert(mid)
|
||
if a and "text" in a:
|
||
text = a["text"]
|
||
try:
|
||
result = self.api.classify(mid, text)
|
||
classes = result.get("classes", [])
|
||
if not classes:
|
||
print(f" [!] {mid}: пустой результат")
|
||
else:
|
||
self._check_assert(mid, result)
|
||
print(f" >> {mid}")
|
||
for c in classes[:5]:
|
||
cl = c.get("class", "?")
|
||
conf = c.get("confidence", 0)
|
||
print(f" class='{cl}' conf={conf:.3f}")
|
||
if len(classes) > 5:
|
||
print(f" ... ({len(classes)} classes)")
|
||
except AssertionError:
|
||
raise
|
||
except Exception as e:
|
||
errors.append(f" {mid}: {e}")
|
||
if errors:
|
||
self.fail("Ошибки internal:\n" + "\n".join(errors))
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="SMC model autotest")
|
||
parser.add_argument("--host", default=DEFAULT_HOST, help=f"Хост (default: {DEFAULT_HOST})")
|
||
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"Порт (default: {DEFAULT_PORT})")
|
||
parser.add_argument("--asserts", default=None, help="JSON файл с ассертами")
|
||
args = parser.parse_args()
|
||
TestSmcModels.args = args
|
||
|
||
suite = unittest.TestLoader().loadTestsFromTestCase(TestSmcModels)
|
||
runner = unittest.TextTestRunner(verbosity=2)
|
||
result = runner.run(suite)
|
||
sys.exit(0 if result.wasSuccessful() else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|