proj/test-memory: Add getline() and getdelim()

proj/server/src/intercept/__init__.py

@@ -1,16 +1,26 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-from typing import Optional, TypedDict, NotRequired, BinaryIO
+from typing import Optional, TypedDict, NamedTuple, NotRequired, BinaryIO
 from socketserver import UnixStreamServer, StreamRequestHandler, ThreadingMixIn
 import os
 import re
 
 
 type Pointer = int
-type PointerTo[T] = tuple[Pointer, T]
-type Constant = tuple[int, str]
-type Flags = tuple[int, list[str]]
+
+class PointerTo[T](NamedTuple):
+    ptr: Pointer
+    target: T
+
+class Constant(NamedTuple):
+    raw: int
+    name: str
+
+class Flags(NamedTuple):
+    bitfield: int
+    flags: list[str]
+
 StructTimeSpec = TypedDict('StructTimeSpec', {'tv_sec': int, 'tv_nsec': int})
 StructSigAction = TypedDict('StructSigAction', {'sa_flags': Flags, 'sa_handler': NotRequired[Pointer], 'sa_sigaction': NotRequired[Pointer], 'sa_mask': list[str]})
 StructSockAddr = TypedDict('StructSockAddr', {'sa_family': Constant, 'sa_data': NotRequired[bytes],
@@ -166,7 +176,7 @@ class Parser:
         if m:
             idx = len(m.group(0)) - 1
             s, i = Parser.parse_str(argument[idx:])
-            idx = i
+            idx += i
             if idx < len(argument) and argument[idx] in ',;':
                 idx += 1
             return s, idx
@@ -207,7 +217,7 @@ class Parser:
             idx += i
             if idx < len(argument) and argument[idx] == ',':
                 idx += 1
-            return (val, list(l)), idx
+            return PointerTo(val, list(l)), idx
         elif argument[idx] == '|':
             m = re.match(r'^[| A-Za-z0-9_]*', argument[idx:])
             flags = m.group(0)
@@ -217,20 +227,20 @@ class Parser:
             if idx < len(argument) and argument[idx] == ',':
                 idx += 1
             flags = [f.strip() for f in flags[1:-1].split('|') if len(f.strip()) > 0]
-            return (val, flags), idx
+            return Flags(val, flags), idx
         elif argument[idx] == '"':
             s, i = Parser.parse_str(argument[idx:])
             idx += i
             if idx < len(argument) and argument[idx] == ',':
                 idx += 1
-            return (val, s), idx
+            return PointerTo(val, s), idx
         elif argument[idx] == '{':
             idx += 1
             l, i = Parser.parse_args(argument[idx:], named=True)
             idx += i
             if idx < len(argument) and argument[idx] == ',':
                 idx += 1
-            return (val, list(l)), idx
+            return PointerTo(val, l), idx
         else:
             m = re.match(r'[A-Z0-9_]+', argument[idx:])
             if not m:
@@ -239,7 +249,7 @@ class Parser:
             idx += len(value)
             if idx < len(argument) and argument[idx] == ',':
                 idx += 1
-            return (val, value), idx
+            return Constant(val, value), idx
 
     @staticmethod
     def parse_args(arguments: str, named: bool = False, ret: bool = False) -> tuple[tuple or dict, int]:
@@ -594,12 +604,12 @@ class Parser:
     def before_sendto(self, sockfd: int, buf: PointerTo[bytes], length: int, flags: Flags, dest_addr: PointerTo[StructSockAddr], addrlen: int) -> str:
         raise NotImplementedError()
     def after_sendto(self, sockfd: int, buf: PointerTo[bytes], length: int, flags: Flags, dest_addr: PointerTo[StructSockAddr], addrlen: int,
-                   ret_value: int, errno: str = None) -> None:
+                     ret_value: int, errno: str = None) -> None:
         raise NotImplementedError()
     def before_sendmsg(self, sockfd: int, message: StructMsgHdr, flags: Flags) -> str:
         raise NotImplementedError()
     def after_sendmsg(self, sockfd: int, message: StructMsgHdr, flags: Flags,
-                   ret_value: int, errno: str = None) -> None:
+                      ret_value: int, errno: str = None) -> None:
         raise NotImplementedError()
     def before_recv(self, sockfd: int, buf_ptr: Pointer, size: int, flags: Flags) -> str:
         raise NotImplementedError()
@@ -609,12 +619,22 @@ class Parser:
     def before_recvfrom(self, sockfd: int, buf_ptr: Pointer, size: int, flags: Flags, src_addr_ptr: Pointer, addrlen_ptr: Pointer) -> str:
         raise NotImplementedError()
     def after_recvfrom(self, sockfd: int, buf_ptr: Pointer, size: int, flags: Flags, src_addr_ptr: Pointer, addrlen_ptr: Pointer,
-                   ret_value: int, errno: str = None, buf: PointerTo[bytes] = None, src_addr: PointerTo[StructSockAddr] = None, addrlen: int = None) -> None:
+                      ret_value: int, errno: str = None, buf: PointerTo[bytes] = None, src_addr: PointerTo[StructSockAddr] = None, addrlen: int = None) -> None:
         raise NotImplementedError()
     def before_recvmsg(self, sockfd: int, message_ptr: Pointer, flags: Flags) -> str:
         raise NotImplementedError()
     def after_recvmsg(self, sockfd: int, message_ptr: Pointer, flags: Flags,
-                   ret_value: int, errno: str = None, message: PointerTo[StructMsgHdr] = None) -> None:
+                      ret_value: int, errno: str = None, message: PointerTo[StructMsgHdr] = None) -> None:
+        raise NotImplementedError()
+    def before_getline(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], stream: Pointer) -> str:
+        raise NotImplementedError()
+    def after_getline(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], stream: Pointer,
+                      ret_value: int, errno: str = None, n: int = None, line: PointerTo[bytes] = None) -> None:
+        raise NotImplementedError()
+    def before_getdelim(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], delim: int, stream: Pointer) -> str:
+        raise NotImplementedError()
+    def after_getdelim(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], delim: int, stream: Pointer,
+                       ret_value: int, errno: str = None, n: int = None, line: PointerTo[bytes] = None) -> None:
         raise NotImplementedError()