local-coding-assistant/test_learning_progress.py at main · M1chlCZ/local-coding-assistant · GitHub

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
91
92
93
94
95
96
97
98
99
100
101
102
"""Keep worker state distinct from finished experiment and benchmark progress."""
import copy
import importlib
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest


class LearningProgressTests(unittest.TestCase):
    def view(self, value, state, confirmation=None):
        self.assertIsNotNone(importlib.util.find_spec('learning_progress'),
                             'Read-only continuous progress view is missing')
        return importlib.import_module('learning_progress').progress_view(value, state, confirmation)

    def setUp(self):
        self.value = {'status': 'running', 'child_status': 'completed', 'phase': 'collect',
            'detail': 'Active time limit reached; progress retained', 'active_seconds': 43202,
            'confirmation': {'tasks': 20}, 'benchmark': {'progress': None}}
        self.state = {'auxiliary_active': True, 'detail': 'Completing reserved matched confirmation checks'}

    def test_completed_experiment_displays_live_reserved_checks(self):
        result = self.view(self.value, self.state,
            {'status': 'running', 'mode': 'base', 'completed': 12, 'total': 20, 'task': 'fresh-task'})
        self.assertEqual(result['status'], 'running')
        self.assertEqual(result['phase'], 'confirmation')
        self.assertEqual(result['confirmation_progress']['completed'], 12)
        self.assertNotIn('time limit', result['detail'])
        self.assertEqual(result['active_seconds'], 43202)

    def test_reference_audit_is_visible_before_first_model_task(self):
        result = self.view(self.value, self.state)
        self.assertEqual(result['confirmation_progress']['phase'], 'reference audit')
        self.assertEqual(result['confirmation_progress']['total'], 20)

    def test_read_only_view_does_not_modify_worker_snapshot(self):
        original = copy.deepcopy(self.value)
        self.view(self.value, self.state, {'mode': 'baseline', 'completed': 1, 'total': 20})
        self.assertEqual(self.value, original)

    def test_pause_and_stop_are_preserved_during_handoff(self):
        for status in ('pausing', 'paused', 'stopping', 'stopped'):
            value = {**self.value, 'status': status}
            result = self.view(value, self.state, {'mode': 'base', 'completed': 12, 'total': 20})
            self.assertEqual(result['status'], status)

    def test_full_benchmark_displays_its_own_phase(self):
        value = {**self.value, 'benchmark': {'progress': {'status': 'running', 'phase': 'adapter',
            'completed': 42, 'total': 164}}}
        result = self.view(value, {**self.state, 'detail': 'Scheduled HumanEval comparison'})
        self.assertEqual(result['phase'], 'benchmark')
        self.assertNotIn('time limit', result['detail'])

    def test_real_failure_remains_visible_with_its_actual_reason(self):
        value = {**self.value, 'status': 'blocked'}
        reason = 'Confirmation server failed to start'
        result = self.view(value, {'auxiliary_active': False, 'detail': reason})
        self.assertEqual(result['status'], 'blocked')
        self.assertEqual(result['detail'], reason)

    def test_saved_benchmark_activity_is_read_only_and_counts_both_models(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory)
            (path/'binding.json').write_text(json.dumps({'languages': ['python', 'go']}))
            for name, rows in {'base-python': [{'id': 'p1'}, {'id': 'p2'}],
                               'base-go': [{'id': 'g1'}, {'id': 'g2'}],
                               'adapter-python': [{'id': 'p1'}, {'id': 'p2'}],
                               'adapter-go': [{'id': 'g1', 'passed': False}]}.items():
                (path/(name+'.json')).write_text(json.dumps(rows))
            before = {p.name: p.read_bytes() for p in path.iterdir()}
            module = importlib.import_module('learning_progress')
            self.assertTrue(hasattr(module, 'benchmark_activity'), 'Saved benchmark activity is missing')
            result = module.benchmark_activity(path, {'phase': 'adapter', 'language': 'go'})
            self.assertEqual(result['stage'], 4)
            self.assertEqual(result['stages'], 4)
            self.assertEqual(result['overall_completed'], 7)
            self.assertEqual(result['overall_total'], 8)
            self.assertEqual(result['last_result'], {'task': 'g1', 'passed': False})
            self.assertEqual({p.name: p.read_bytes() for p in path.iterdir()}, before)

    def test_incomplete_base_reports_do_not_invent_an_overall_total(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory)
            (path/'binding.json').write_text(json.dumps({'languages': ['python', 'go']}))
            (path/'base-python.json').write_text(json.dumps([{'id': 'p1', 'passed': True}]))
            module = importlib.import_module('learning_progress')
            self.assertTrue(hasattr(module, 'benchmark_activity'), 'Saved benchmark activity is missing')
            result = module.benchmark_activity(path, {'phase': 'base', 'language': 'python'})
            self.assertEqual(result['stage'], 1)
            self.assertEqual(result['overall_completed'], 1)
            self.assertNotIn('overall_total', result)

    def test_old_single_language_benchmark_still_displays_its_own_progress(self):
        with tempfile.TemporaryDirectory() as directory:
            module = importlib.import_module('learning_progress')
            self.assertTrue(hasattr(module, 'benchmark_activity'), 'Saved benchmark activity is missing')
            self.assertEqual(module.benchmark_activity(Path(directory), {'phase': 'base'}), {})


if __name__ == '__main__':
    unittest.main()