xc_agent_platform_demo/tests/test_pending_pool.py

import tempfile
import unittest
from pathlib import Path
from src.pending_pool import build_ordered_candidates, parse_pending_txt


class TestPendingPool(unittest.TestCase):
    def test_order_filters_and_first_gpu(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            md, txt, dnr, banned = [root / n for n in ('pending.md', 'pending.txt', 'dnr.md', 'banned.md')]
            md.write_text('''## Section B
| 1 | `org/cold` | 50 | | | 
## Section A
| 1 | `org/good` | 曦望 | 壁砺, 天垓150 | — |
| 2 | `org/exact` | 曦望 | 壁砺 | — |
| 3 | `org/Heretic-model` | 曦望 | 壁砺 | — |
| 4 | `org/OLMo-2-model` | 曦望 | 壁砺 | — |
''')
            dnr.write_text('| 1 | `org/exact` | fail |\n## Uniqueness collisions\n- `org/collision`\n## Overnight\n- `org/overnight`')
            banned.write_text('`abliterated` `heretic` `uncensored` `unsensored`')
            rows = build_ordered_candidates(md, txt, dnr, banned)
            self.assertEqual([r['section'] for r in rows], ['A'] * 4 + ['B'])
            self.assertEqual(rows[0]['try_next_gpus'][0], '壁砺')
            self.assertEqual(rows[1]['skip_reason'], 'do-not-resubmit')
            self.assertEqual(rows[2]['status'], 'skipped')
            self.assertIn('family', rows[3]['skip_reason'])
            self.assertTrue(rows[-1]['needs_human'])
            txt.write_text('org/b\t#B cold\norg/a\t#A done:曦望; still try:壁砺+天垓150\norg/c\\t#A try:天垓150\n')
            self.assertEqual(parse_pending_txt(txt)[1]['try_next_gpus'], ['壁砺', '天垓150'])
            fallback = build_ordered_candidates(root/'absent', txt, dnr, banned)
            self.assertEqual([r['model_address'] for r in fallback], ['org/a', 'org/c', 'org/b'])
            self.assertEqual(len(build_ordered_candidates(md, txt, dnr, banned)), 5)

    def test_experience_order_missing_only_and_ties(self):
        from src.pending_pool import experience_order, parse_pending_md
        self.assertEqual(experience_order(['智铠 mrv-100', '天垓100', '曦望', '天垓150', '壁砺']),
                         ['曦望', '壁砺', '天垓150', '天垓100', '智铠 mrv-100'])
        self.assertEqual(experience_order(['天垓150']), ['天垓150'])
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / 'pending.md'
            path.write_text('## Section A\n| 1 | `org/a` | 壁砺:验证失败; 曦望:正在验证 | 天垓150 | no 壁砺 retry |\n## Section B\n| 1 | `org/b` | 0 | 壁砺 |\n')
            from unittest.mock import patch
            import os
            with patch.dict(os.environ, {'PENDING_SECTION': 'A'}):
                rows = build_ordered_candidates(path, dnr_path=Path(directory)/'none', banned_path=Path(directory)/'none')
            self.assertEqual(len(rows), 1)
            self.assertEqual(rows[0]['try_next_gpus'], ['天垓150'])
            self.assertEqual(rows[0]['perf_order'], 'unset')

    def test_foothold_evidence_and_stable_verified_priority(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            md = root / 'pending.md'
            md.write_text('## Section B\n| 1 | `org/b` | 0 | 壁砺 |\n## Section A\n| 1 | `org/unknown` | 曦望 | 壁砺 | 已验证成功 |\n| 2 | `org/flight` | 验证失败; 正在验证 | 壁砺 | — |\n| 3 | `org/verified` | 他人:已验证成功; 正在验证 | 壁砺 | — |\n| 4 | `org/fail` | 验证失败 | 曦望 | — |\n| 5 | `org/pass` | 既有:验证通过 | 曦望 | — |\n| 6 | `org/queue` | 验证排队中 | 壁砺 | — |\n| 7 | `org/progress` | 验证进行中 | 壁砺 | — |\n')
            rows = build_ordered_candidates(md, dnr_path=root/'none', banned_path=root/'none')
            self.assertEqual([r['model_address'] for r in rows],
                             ['org/verified', 'org/pass', 'org/unknown', 'org/flight',
                              'org/fail', 'org/queue', 'org/progress', 'org/b'])
            self.assertEqual([r['foothold_class'] for r in rows],
                             ['verified_reuse', 'verified_reuse', 'unknown', 'in_flight',
                              'failed_foothold', 'queued_only', 'in_flight', 'unknown'])
            for row in rows:
                self.assertEqual(row['evidence_source'], 'GAP' if row['foothold_class'] == 'unknown' else 'pending-foothold')
            self.assertEqual(rows[0]['foothold_text'], '他人:已验证成功; 正在验证')
            txt = root / 'pending.txt'
            txt.write_text('org/a\t#A done:曦望; try:壁砺\norg/v\t#A 验证通过; try:壁砺\n')
            parsed = parse_pending_txt(txt)
            self.assertEqual(parsed[0]['foothold_class'], 'unknown')
            self.assertEqual(parsed[0]['evidence_source'], 'GAP')
            self.assertEqual(parsed[1]['foothold_class'], 'verified_reuse')

    def test_no_download_marker_without_inventing(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            md = root / 'pending.md'
            md.write_text('## Section A\n| 1 | `org/plain` | 曦望:验证排队中 | 壁砺 | — |\n| 2 | `org/cached` | 壁砺:已下载; 曦望:免下载 | 天垓150 | — |\n| 3 | `org/verified` | 壁砺:已验证成功 | 天垓150 | — |\n')
            rows = build_ordered_candidates(md, dnr_path=root/'none', banned_path=root/'none')
            by = {r['model_address']: r for r in rows}
            self.assertFalse(by['org/plain']['no_download'])
            self.assertEqual(by['org/plain']['evidence_source'], 'pending-foothold')
            self.assertTrue(by['org/cached']['no_download'])
            self.assertEqual(by['org/cached']['evidence_source'], 'pending-foothold')
            self.assertFalse(by['org/verified']['no_download'])  # verified alone does not invent no_download
            self.assertEqual([r['model_address'] for r in rows],
                             ['org/cached', 'org/verified', 'org/plain'])