python support @broken annotation

README.md

@@ -381,6 +381,36 @@ def throw_wrong():
     empty['xxx']
 ```
 
+#### testa.broken
+
+We also can mark cases as broken ones for some reasons.
+
+```python
+@testa.broken(reason = 'this is a broken case.')
+@testa.is_(expect=3)
+def broken_addition():
+    return 2 + 2
+
+@testa.broken(reason = 'this is a broken case.')
+@testa.eq(testbench=multiple_tb, oracle=multiple_oracle)
+def broken_multiple(x, y):
+    return x + y
+
+@testa.broken(reason = 'this is a broken case.')
+@testa.verify(testbench=gcd_tb, trial=gcd)
+def broken_verifier(result, a, b):
+    if a == 0 and b == 0:
+        return 'undefined on (0, 0)'
+    if a == 1:
+        return 'fake error on a=1'
+
+@testa.broken(reason = 'this is a broken case.')
+@testa.throw(throw=IOError)
+def broken_throw():
+    empty = {}
+    empty['xxx']
+```
+
 ## How to run tests
 
 Here are some examples.

python/testa.py

@@ -5,6 +5,19 @@
 
 fixtures = {}
 
+class Case:
+    def __init__(self, case_f) -> None:
+        self.case_f = case_f
+        self.is_broken = False
+        self.broken_reason = None
+
+def broken(**kws):
+    def go(cs):
+        cs.is_broken = True
+        assert 'reason' in kws, 'broken case requires a reason'
+        cs.broken_reason = kws['reason']
+    return go
+
 def _is_testbench(trial_f, expect):
     actual = trial_f()
     if actual != expect:
@@ -15,7 +28,9 @@ def go(trial_f):
         global fixtures
         casename = trial_f.__name__
         expect = kws['expect']
-        fixtures[casename] = lambda: _is_testbench(trial_f, expect)
+        res = Case(lambda: _is_testbench(trial_f, expect))
+        fixtures[casename] = res
+        return res
     return go
 
 def _eq_case_f(trial_f, oracle_f, *args):
@@ -34,7 +49,9 @@ def go(trial_f):
         tb = kws['testbench']
         oracle_f = kws['oracle']
         case_f = lambda *args: _eq_case_f(trial_f, oracle_f, *args)
-        fixtures[casename] = lambda: tb(case_f)
+        res = Case(lambda: tb(case_f))
+        fixtures[casename] = res
+        return res
     return go
 
 def _verify_case_f(trial_f, verifier, *args):
@@ -50,7 +67,9 @@ def go(verifier_f):
         tb = kws['testbench']
         trial_f = kws['trial']
         case_f = lambda *args: _verify_case_f(trial_f, verifier_f, *args)
-        fixtures[casename] = lambda: tb(case_f)
+        res = Case(lambda: tb(case_f))
+        fixtures[casename] = res
+        return res
     return go
 
 def _throws_tb(trial_f, expect_except):
@@ -66,7 +85,9 @@ def go(trial_f):
         global fixtures
         casename = trial_f.__name__
         expect_throw = kws['throw']
-        fixtures[casename] = lambda: _throws_tb(trial_f, expect_throw)
+        res = Case(lambda: _throws_tb(trial_f, expect_throw))
+        fixtures[casename] = res
+        return res
     return go
 
 def _parse_args():
@@ -84,14 +105,20 @@ def _parse_args():
         return 'case', args.casename
 
 def main():
+    global fixtures
     action, case = _parse_args()
     if action == 'list':
         res = [{'name': x} for x in sorted(fixtures.keys())]
+        for x in res:
+            cs = fixtures[x['name']]
+            if cs.is_broken:
+                x['broken'] = True
+                x['broken_reason'] = cs.broken_reason
         print(json.dumps(res, sort_keys=True, indent=2))
         sys.exit(0)
     elif action == 'case':
         try:
-            t = fixtures[case]
+            t = fixtures[case].case_f
         except KeyError:
             print('unknown case:', case)
             sys.exit(1)

python/testa_test.py

@@ -8,6 +8,11 @@ def correct_addition():
 def wrong_addition():
     return 2 + 2
 
+@testa.broken(reason = 'this is a broken case.')
+@testa.is_(expect=3)
+def broken_addition():
+    return 2 + 2
+
 def multiple_tb(case_f):
     for i in range(5):
         for j in range(5):
@@ -26,6 +31,12 @@ def correct_multiple(x, y):
 def wrong_multiple(x, y):
     return x + y
 
+@testa.broken(reason = 'this is a broken case.')
+@testa.eq(testbench=multiple_tb, oracle=multiple_oracle)
+def broken_multiple(x, y):
+    return x + y
+
+
 def gcd_tb(case_f):
     for a in range(13):
         for b in range(13):
@@ -57,6 +68,15 @@ def wrong_verifier(result, a, b):
     if a == 1:
         return 'fake error on a=1'
 
+@testa.broken(reason = 'this is a broken case.')
+@testa.verify(testbench=gcd_tb, trial=gcd)
+def broken_verifier(result, a, b):
+    if a == 0 and b == 0:
+        return 'undefined on (0, 0)'
+    if a == 1:
+        return 'fake error on a=1'
+
+
 @testa.throw(throw=KeyError)
 def throw_correct():
     empty = {}
@@ -67,5 +87,11 @@ def throw_wrong():
     empty = {}
     empty['xxx']
 
+@testa.broken(reason = 'this is a broken case.')
+@testa.throw(throw=IOError)
+def broken_throw():
+    empty = {}
+    empty['xxx']
+
 if __name__ == '__main__':
     testa.main()

runtests.py

@@ -93,6 +93,7 @@ def readLangCfg(fn):
 kError = 'Error'
 kTimeout = 'Timeout'
 kCancel = 'Cancel'
+kSkip = 'Skip'
 
 gCancelled = False
 
@@ -108,6 +109,10 @@ def work(opts, qin, qout):
             args = shlex.split(cs['execute'])
             kws = {}
             with open(cs['stdout'], 'wb') as stdout, open(cs['stderr'], 'wb') as stderr:
+                if cs.get('broken', False):
+                    stdout.write(cs['broken-reason'].encode())
+                    qout.put([kSkip, cs['name'], cs])
+                    break
                 kws['stdout'] = stdout
                 kws['stderr'] = stderr
                 kws['check'] = True
@@ -189,12 +194,17 @@ def collectCases(opts, langs, reqQ, resQ):
             cs = json.load(f)
         lang = findMatchLanguage(exe, langs)
         for c in cs:
-            cases.append({
+            x = {
                 'name': '%s/%s' % (exe, c['name']),
+                'broken': c.get('broken', False),
                 'execute': lang['execute'] % {'prog': op.abspath(exe), 'arg': c['name']},
                 'cwd': res[2]['cwd'],
                 'stdout': op.join(opts.dir, exe, '%s.out' % c['name']),
-                'stderr': op.join(opts.dir, exe, '%s.err' % c['name'])})
+                'stderr': op.join(opts.dir, exe, '%s.err' % c['name']),
+            }
+            if x['broken']:
+                x['broken-reason'] = c['broken_reason']
+            cases.append(x)
     return cases
 
 SUPPRESS_TERMCOLOR_DETECTION = False
@@ -231,23 +241,29 @@ def collectResults(opts, cases, resQ):
     while len(passed) + len(failed) < caseNum:
         res = resQ.get()
         r = res[2].copy()
-        r['duration'] = res[2]['stop'] - res[2]['start']
-        del r['start']
-        del r['stop']
-        if res[0] == kOk:
-            r['result'] = 'PASS'
+        if res[0] == kSkip:
+            r['result'] = 'SKIP'
+            r['duration'] = timedelta()
             passed.append(r)
-            result = colored('pass', 'green')
-        elif res[0] == kError:
-            r['result'] = 'FAILED'
-            failed.append(r)
-            result = colored('fail', 'red')
-        elif res[0] == kTimeout:
-            r['result'] = 'TIMEOUT'
-            failed.append(r)
-            result = colored('kill', 'red')
+            result = colored('skip', 'blue')
         else:
-            error('cancelled')
+            r['duration'] = res[2]['stop'] - res[2]['start']
+            del r['start']
+            del r['stop']
+            if res[0] == kOk:
+                r['result'] = 'PASS'
+                passed.append(r)
+                result = colored('pass', 'green')
+            elif res[0] == kError:
+                r['result'] = 'FAILED'
+                failed.append(r)
+                result = colored('fail', 'red')
+            elif res[0] == kTimeout:
+                r['result'] = 'TIMEOUT'
+                failed.append(r)
+                result = colored('kill', 'red')
+            else:
+                error('cancelled')
         print('%d/%d %s: %s costs %.6f secs' % (
             len(passed) + len(failed), caseNum,
             result,