xc_agent_platform_demo/tests/test_live_queue.py

import json
import os
import unittest
from unittest.mock import patch, MagicMock
from src.live_queue import read_queue


class TestLiveQueue(unittest.TestCase):
    def response(self, body):
        r = MagicMock()
        r.__enter__.return_value = r
        r.status = 200
        r.read.return_value = json.dumps(body).encode()
        return r

    def test_pagination_and_status(self):
        pages = [self.response({'code': 0, 'data': {'total': '3', 'records': [
            {'status': 'waiting', 'modelId': 'org/a', 'gpuType': 'Biren_166m'}, {'status': 'success', 'verifyResult': -1}]}}),
            self.response({'code': 0, 'data': {'total': '3', 'records': [{'status': '等待中'}]}})]
        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', side_effect=pages) as call:
            count, why, models = read_queue()
        self.assertEqual(count, 2)
        self.assertIn(('org/a', 'Biren_166m'), models)
        self.assertIn('current=2', call.call_args.args[0].full_url)

    def test_fail_closed(self):
        for body in ({'code': 0, 'data': {'total': 'bad', 'records': []}}, {'code': 40100}, {'code': 40101}, {'code': 0, 'data': {'total': 1, 'records': []}}, []):
            with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', return_value=self.response(body)):
                self.assertIsNone(read_queue()[0])
        with patch.dict(os.environ, {}, clear=True), patch('urllib.request.urlopen') as call:
            self.assertIsNone(read_queue()[0])
            call.assert_not_called()

    def test_page_limit_and_transport(self):
        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True):
            with patch('urllib.request.urlopen', return_value=self.response({'code': 0, 'data': {'total': 21, 'records': [{'status': 'waiting'}]}})) as call:
                self.assertIsNone(read_queue()[0])
                self.assertEqual(call.call_count, 20)
            with patch('urllib.request.urlopen', side_effect=OSError('synthetic-secret')):
                self.assertNotIn('synthetic-secret', str(read_queue()))

    def test_invalid_json_http_and_auth_header(self):
        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True):
            r = self.response({})
            r.read.return_value = b'invalid synthetic-secret'
            with patch('urllib.request.urlopen', return_value=r):
                self.assertEqual(read_queue()[1], 'malformed_json')
            r.status = 503
            with patch('urllib.request.urlopen', return_value=r):
                self.assertIsNone(read_queue()[0])
            with patch('urllib.request.urlopen', return_value=self.response({'code': 0, 'data': {'total': 0, 'records': []}})) as call:
                self.assertEqual(read_queue()[0], 0)
                req = call.call_args.args[0]
                self.assertEqual(req.get_header('Xc-token'), 'synthetic-secret')
                self.assertEqual(req.method, 'GET')


class TestVerifyStats(unittest.TestCase):
    response = TestLiveQueue.response

    def test_compute_stats_precedence_and_types(self):
        from src.live_queue import compute_stats
        rows = [
            dict(status='waiting', verifyResult=1, stageAndNoviceTaskCredit=100),
            dict(status='等待中', verifyResult=-1),
            dict(status='success', verifyResult=-1, stageAndNoviceTaskCredit=100),
            dict(status='failed', verifyResult=' -1 '),
            dict(status='success'), dict(status='success', verifyResult=True),
            dict(status='success', verifyResult=False),
            dict(status='success', verifyResult='garbage'),
            dict(status='error', verifyResult=None),
            dict(status=' CANCELED ', verifyResult=0),
        ]
        for result, credit in [(1, '5'), ('1', 5.0), (1.0, None), (1, 'abc'), (1, True), (1, 'NaN')]:
            rows.append(dict(status='success', verifyResult=result, stageAndNoviceTaskCredit=credit))
        stats = compute_stats(rows)
        self.assertEqual(stats, dict(total=16, queue=2, verify_pass=6, verify_fail=2,
                                     failed=2, unknown=4, points=10.0))
        self.assertEqual(sum(stats[k] for k in ('queue', 'verify_pass', 'verify_fail', 'failed', 'unknown')), stats['total'])
        self.assertIs(type(stats['points']), float)
        self.assertTrue(all(type(v) is int for k, v in stats.items() if k != 'points'))

    def test_paginated_stats_and_snapshot(self):
        from src.live_queue import read_queue_stats, last_stats
        pages = [self.response({'code': 0, 'data': {'total': 2, 'records': [row]}})
                 for row in [dict(status='success', verifyResult=1, stageAndNoviceTaskCredit='5', body='synthetic-secret'),
                             dict(status='success', verifyResult=-1)]]
        with patch.dict(os.environ, {'XC_TOKEN': 'synthetic-secret'}, clear=True), patch('urllib.request.urlopen', side_effect=pages):
            count, reason, models, stats = read_queue_stats()
        self.assertEqual((count, reason, models), (0, 'ok', set()))
        self.assertEqual(stats['verify_pass'], 1)
        self.assertEqual(stats['verify_fail'], 1)
        self.assertEqual(stats['points'], 5.0)
        snapshot = last_stats()
        self.assertIn('updated_at', snapshot)
        self.assertNotIn('synthetic-secret', json.dumps(snapshot))
        snapshot['total'] = 999
        stats['total'] = 888
        self.assertEqual(last_stats()['total'], 2)
        with patch.dict(os.environ, {}, clear=True):
            self.assertEqual(read_queue_stats(), (None, 'missing/invalid_token', set(), None))
        self.assertEqual(last_stats()['total'], 2)
        with patch('src.live_queue.read_queue_stats', return_value=(3, 'ok', set(), {})):
            self.assertEqual(read_queue(), (3, 'ok', set()))