proj: Enhance ./test-return-values

proj/server/src/intercept/__init__.py

@@ -90,7 +90,7 @@ class Handler(StreamRequestHandler):
     def parse_arg(argument: str) -> tuple[any, int]:
         if argument == '':
             return None, 0
-        m = re.match(r'\s*\(nil\)', argument)
+        m = re.match(r'^\s*\(nil\)\s*(,|$)', argument)
         if m:
             return 0, len(m.group(0))
         m = re.match(r'^\s*(.*?)([,:]|$)', argument)

proj/server/src/intercept/standard.py

@@ -157,9 +157,91 @@ class InterruptedCheckTester(Handler):
             return self.functions[func_name][1]
 
 
-class ReturnValueCheckTester(Handler):
+def get_return_value_check_tester() -> tuple[dict, type[Handler]]:
-    functions: dict[str, list[str]] = {
+    ctx = {
-        fn: [e for e in errors if e not in SKIP_ERRORS and e not in IGNORE_ERRORS]
+        'state': 'init',
-        for fn, errors in FUNCTION_ERRORS.items()
+        'call_sequence': [],
-        if len(set(errors) - set(SKIP_ERRORS) - set(IGNORE_ERRORS)) > 0
+        'next': None,
+        'last_error': None,
+        'results': {},
     }
+
+    class ReturnValueCheckTester(Handler):
+        functions: dict[str, list[str]] = {
+            fn: [e for e in errors if e not in SKIP_ERRORS and e not in IGNORE_ERRORS]
+            for fn, errors in FUNCTION_ERRORS.items()
+            if len(set(errors) - set(SKIP_ERRORS) - set(IGNORE_ERRORS)) > 0
+        }
+        context: dict
+        num: int = 0
+
+        @property
+        def call_sequence(self) -> list[str]:
+            return self.context['call_sequence']
+
+        @property
+        def is_init(self) -> bool:
+            return self.context['state'] == 'init'
+
+        @property
+        def is_testing(self) -> bool:
+            return self.context['state'] == 'testing'
+
+        @property
+        def is_waiting(self) -> bool:
+            return self.context['state'] == 'waiting'
+
+        @property
+        def is_failed(self) -> bool:
+            return self.context['state'] == 'failed'
+
+        def next(self, last: Optional[tuple[int, Optional[str]]]) -> tuple[int, Optional[str]]:
+            if last is None:
+                if len(self.call_sequence) == 0:
+                    return 0, None
+                err = self.functions.get(self.call_sequence[0], [None])[0]
+                return (0, err) if err else self.next((0, err))
+            last_fn, last_er = last
+            if last_er is None:
+                if last_fn + 1 >= len(self.call_sequence):
+                    return last_fn + 1, None
+                err = self.functions.get(self.call_sequence[last_fn + 1], [None])[0]
+                return (last_fn + 1, err) if err else self.next((last_fn + 1, err))
+            errors = self.functions.get(self.call_sequence[last_fn], [])
+            i = errors.index(last_er)
+            return (last_fn, errors[i + 1]) if i + 1 < len(errors) else self.next((last_fn, None))
+
+        def before(self) -> None:
+            self.context = ctx
+            if self.is_waiting or self.is_failed:
+                self.context['state'] = 'testing'
+
+        def after(self) -> None:
+            if self.is_init:
+                self.context['state'] = 'testing'
+                self.context['next'] = self.next(None)
+            elif self.is_testing and self.context['next'][0] > len(self.context['call_sequence']):
+                self.context['state'] = 'finished'
+            elif self.is_waiting:
+                self.context['results'][(self.num - 1, self.context['last_error'])] = 'passed'
+
+        def before_fallback(self, func_name: str, *args) -> str:
+            if self.is_init:
+                self.call_sequence.append(func_name)
+                return 'ok'
+            elif self.is_waiting or self.is_failed:
+                print(func_name)
+                if func_name not in ('malloc', 'free', 'close', 'exit'):
+                    self.context['state'] = 'failed'
+                    self.context['results'][(self.num - 1, self.context['last_error'])] = 'failed'
+                return 'ok'
+            self.num += 1
+            nxt = self.context['next']
+            if self.num - 1 != nxt[0] or nxt[1] is None:
+                return 'ok'
+            self.context['next'] = self.next(nxt)
+            self.context['state'] = 'waiting' if self.context['next'][1] else 'finished'
+            self.context['last_error'] = nxt[1]
+            return 'fail ' + nxt[1]
+
+    return ctx, ReturnValueCheckTester

proj/server/src/test-return-values

@@ -2,6 +2,7 @@
 # -*- coding: utf-8 -*-
 
 import os
+import sys
 import argparse
 import threading
 import subprocess
@@ -10,18 +11,43 @@ import intercept
 import intercept.standard
 
 
-def socket_thread(socket: str) -> None:
+def socket_thread(socket: str, handler: type[intercept.Handler]) -> None:
-    intercept.intercept(socket, intercept.standard.ReturnValueCheckTester)
+    intercept.intercept(socket, handler)
 
 
 def main() -> None:
     parser = argparse.ArgumentParser()
+    parser.add_argument('-i', '--stdin')
     args, extra = parser.parse_known_args()
+    if len(extra) > 0 and extra[0] == '--':
+        extra.pop(0)
+
+    stdin = open(args.stdin) if args.stdin else None
+
     socket_name = f'/tmp/intercept.return-values.{os.getpid()}.sock'
-    t1 = threading.Thread(target=socket_thread, args=(socket_name,))
+    ctx, handler = intercept.standard.get_return_value_check_tester()
+    t1 = threading.Thread(target=socket_thread, args=(socket_name, handler))
     t1.daemon = True
     t1.start()
-    subprocess.run(extra, env={'LD_PRELOAD': os.getcwd() + '/../../intercept/intercept.so', 'INTERCEPT': 'unix:' + socket_name})
+    while ctx['state'] != 'finished':
+        if stdin:
+            stdin.seek(0, 0)
+        subprocess.run(extra, stdin=stdin, env={'LD_PRELOAD': os.getcwd() + '/../../intercept/intercept.so', 'INTERCEPT': 'unix:' + socket_name})
+    for i, name in enumerate(ctx['call_sequence']):
+        errors = [r[1] for r in ctx['results'] if r[0] == i]
+        results = [ctx['results'][(i, e)] for e in errors]
+        res = '-' if len(results) == 0 else 'passed' if all(r == 'passed' for r in results) else 'failed'
+        esc, unesc = '', ''
+        errs = [e for e in errors if ctx['results'][(i, e)] == res]
+        errs = f' ({", ".join(errs)})' if len(errs) > 0 else ''
+        if sys.stdout.isatty():
+            if res == '-':
+              esc, unesc = '\x1B[90m', '\x1B[0m'
+            elif res == 'failed':
+                res = '\x1B[31;1mfailed\x1B[0m'
+            elif res == 'passed':
+                res = '\x1B[32;1mpassed\x1B[0m'
+        print(f'{esc}{i:3} {name:16} {res}{unesc}{errs}')
 
 
 if __name__ == '__main__':

proj/test1/src/main.c

@@ -17,7 +17,7 @@ void do_something(void) {
     void *mem = malloc(123);
     if (mem == NULL) {
         fprintf(stderr, "Unable to malloc: %s\n", strerror(errno));
-        return;
+        exit(1);
     }
     printf("%p\n", mem);
     free(mem);