88 lines
4.7 KiB
Python
88 lines
4.7 KiB
Python
"""Run: python -B -m unittest discover -s tests -v (uses a temporary model store)."""
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
_temporary = tempfile.TemporaryDirectory()
|
|
os.environ['SLEARN_MODEL_STORE'] = _temporary.name
|
|
|
|
import model_registry as registry
|
|
from model_service import analisis_teks
|
|
from main import TrainingRequest, training
|
|
from fastapi import BackgroundTasks, HTTPException
|
|
|
|
|
|
class RetrainingTests(unittest.TestCase):
|
|
def test_full_lifecycle(self):
|
|
baseline = registry.bundle('v1')
|
|
frame, train, test = registry.seed()
|
|
self.assertEqual((len(train), len(test)), (456, 115))
|
|
self.assertAlmostEqual(baseline[2]['akurasi'], 0.9304347826086956)
|
|
original_hash = hashlib.sha256((registry.ARTIFACTS / 'model_logistic_regression.joblib').read_bytes()).hexdigest()
|
|
text = 'Aplikasi bagus sangat membantu belajar dengan materi menarik terbaru sekali'
|
|
before = analisis_teks(text)
|
|
queued = registry.queue('contoh.csv')
|
|
with self.assertRaises(ValueError):
|
|
registry.queue('duplikat-proses.csv')
|
|
registry.train_job(queued, [text, text, test.iloc[0].komentar, '', '123 !!!', 'zzqqxxvv'])
|
|
state = registry.history()
|
|
self.assertEqual(state['jobs'][0]['status'], 'completed', state['jobs'][0]['message'])
|
|
self.assertEqual(state['active_version'], 'v1')
|
|
self.assertEqual(state['versions'][0]['jumlah_data_training'], 457)
|
|
self.assertEqual(state['versions'][0]['skipped'], {'duplicate': 2, 'empty': 2, 'unknown': 1})
|
|
self.assertEqual(state['versions'][0]['teacher_version'], 'v1')
|
|
self.assertEqual(state['versions'][0]['jumlah_data_testing'], 115)
|
|
registry.activate('v2')
|
|
self.assertEqual(analisis_teks(text)['model_version'], 'v2')
|
|
# Old snapshots remain internally consistent during activation.
|
|
self.assertEqual(analisis_teks(text, snapshot=baseline), before)
|
|
registry._cache.clear()
|
|
self.assertEqual(registry.bundle()[2]['version'], 'v2')
|
|
registry.activate('v1')
|
|
self.assertEqual(analisis_teks(text), before)
|
|
self.assertEqual(original_hash, hashlib.sha256((registry.ARTIFACTS / 'model_logistic_regression.joblib').read_bytes()).hexdigest())
|
|
# Training after rollback still includes all previously committed additions.
|
|
job = registry.queue('tambahan.csv')
|
|
registry.train_job(job, ['Website lambat sering error saat mengumpulkan tugas malam terbaru'])
|
|
self.assertEqual(registry.history()['jobs'][0]['status'], 'completed')
|
|
self.assertEqual(registry.history()['versions'][0]['jumlah_data_training'], 458)
|
|
with self.assertRaises(ValueError):
|
|
registry.activate('../../invalid')
|
|
self.assertEqual(registry.history()['active_version'], 'v1')
|
|
failed = registry.queue('hanya-data-lama.csv')
|
|
registry.train_job(failed, [text, test.iloc[1].komentar])
|
|
state = registry.history()
|
|
self.assertEqual(state['jobs'][0]['status'], 'skipped')
|
|
self.assertEqual(len(state['versions']), 3)
|
|
with registry.connect() as db:
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM additions').fetchone()[0], 2)
|
|
failed_fit = registry.queue('gagal-fit.csv')
|
|
with patch('sklearn.linear_model.LogisticRegression.fit', side_effect=RuntimeError('simulated failure')):
|
|
registry.train_job(failed_fit, ['Materi pembelajaran bagus membantu mahasiswa mengerjakan latihan unik'])
|
|
self.assertEqual(registry.history()['jobs'][0]['status'], 'failed')
|
|
self.assertEqual(len(registry.history()['versions']), 3)
|
|
with registry.connect() as db:
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM additions').fetchone()[0], 2)
|
|
registry.queue('terputus.csv')
|
|
registry.initialize()
|
|
self.assertEqual(registry.history()['jobs'][0]['status'], 'failed')
|
|
|
|
def test_request_validation(self):
|
|
for reviews in [[' '], ['x' * 10001]]:
|
|
with self.assertRaises(HTTPException):
|
|
training(TrainingRequest(file_name='invalid.csv', reviews=reviews, selected_column='ulasan', confirmed_review_column=True), BackgroundTasks())
|
|
|
|
def test_predict_route_uses_active_model(self):
|
|
from main import predict, TextRequest
|
|
with patch('main.save_single_prediction', return_value=123):
|
|
response = predict(TextRequest(text='Aplikasi bagus membantu belajar'))
|
|
self.assertEqual(response['data']['prediction_id'], 123)
|
|
self.assertEqual(response['data']['model_version'], registry.history()['active_version'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|