aboutsummaryrefslogtreecommitdiff
path: root/src/test
diff options
context:
space:
mode:
Diffstat (limited to 'src/test')
-rw-r--r--src/test/__init__.py13
-rw-r--r--src/test/error.py62
-rw-r--r--src/test/philo.py194
-rw-r--r--src/test/test.py148
4 files changed, 0 insertions, 417 deletions
diff --git a/src/test/__init__.py b/src/test/__init__.py
deleted file mode 100644
index 4cc06f8..0000000
--- a/src/test/__init__.py
+++ /dev/null
@@ -1,13 +0,0 @@
-# ############################################################################ #
-# #
-# ::: :::::::: #
-# __init__.py :+: :+: :+: #
-# +:+ +:+ +:+ #
-# By: charles <me@cacharle.xyz> +#+ +:+ +#+ #
-# +#+#+#+#+#+ +#+ #
-# Created: 2020/09/27 13:00:31 by charles #+# #+# #
-# Updated: 2020/09/27 17:56:09 by charles ### ########.fr #
-# #
-# ############################################################################ #
-
-from test.test import Test # noqa: F401
diff --git a/src/test/error.py b/src/test/error.py
deleted file mode 100644
index 51c3f7b..0000000
--- a/src/test/error.py
+++ /dev/null
@@ -1,62 +0,0 @@
-# ############################################################################ #
-# #
-# ::: :::::::: #
-# error.py :+: :+: :+: #
-# +:+ +:+ +:+ #
-# By: cacharle <me@cacharle.xyz> +#+ +:+ +#+ #
-# +#+#+#+#+#+ +#+ #
-# Created: 2020/09/29 09:09:31 by cacharle #+# #+# #
-# Updated: 2020/09/29 11:14:08 by cacharle ### ########.fr #
-# #
-# ############################################################################ #
-
-class Philo(Exception):
- pass
-
-
-class ShouldFail(Philo):
- def __init__(self, msg: str):
- self._msg = msg
- Philo.__init__(self)
-
- @property
- def full_summary(self):
- return self.summary
-
- @property
- def summary(self):
- return "Should fail: {}".format(self._msg)
-
-
-class Format(Philo):
- def __init__(self, line: str, msg: str):
- self._line = line
- self._msg = msg
- Philo.__init__(self)
-
- @property
- def full_summary(self):
- return """FORMAT ERROR: {}
-{}
-""".format(self._msg, self._line)
-
- @property
- def summary(self):
- return "format: {} {}".format(self._line, self._msg)
-
-
-class Log(Philo):
- def __init__(self, logs: [str], msg: str):
- self._logs = logs
- self._msg = msg
- Philo.__init__(self)
-
- @property
- def full_summary(self):
- return """LOG ERROR: {}
-{}
-""".format(self._msg, '\n'.join([str(l) for l in self._logs]))
-
- @property
- def summary(self):
- return "log: {}".format(self._msg)
diff --git a/src/test/philo.py b/src/test/philo.py
deleted file mode 100644
index 3b49057..0000000
--- a/src/test/philo.py
+++ /dev/null
@@ -1,194 +0,0 @@
-# ############################################################################ #
-# #
-# ::: :::::::: #
-# philo.py :+: :+: :+: #
-# +:+ +:+ +:+ #
-# By: charles <me@cacharle.xyz> +#+ +:+ +#+ #
-# +#+#+#+#+#+ +#+ #
-# Created: 2020/09/27 17:49:41 by charles #+# #+# #
-# Updated: 2020/09/30 15:02:10 by cacharle ### ########.fr #
-# #
-# ############################################################################ #
-
-import re
-import enum
-import itertools
-
-import test.error as error
-from helper import current_ms
-
-class Event(enum.Enum):
- FORK = 1
- EATING = 2
- SLEEPING = 3
- THINKING = 4
- DIED = 5
- NONE = 6
-
- @staticmethod
- def from_string(representation: str) -> Event:
- return {
- "has taken fork": Event.FORK,
- "is thinking": Event.THINKING,
- "is eating": Event.EATING,
- "is sleeping": Event.SLEEPING,
- "died": Event.DIED,
- }[representation]
-
- @staticmethod
- def to_string(event: Event) -> str:
- return {
- Event.FORK: "has taken fork",
- Event.THINKING: "is thinking",
- Event.EATING: "is eating",
- Event.SLEEPING: "is sleeping",
- Event.DIED: "died"
- }[event]
-
-
-
-class Log:
- def __init__(self, line: str, philo_num, start_time):
- match = re.match(
- r"^(?P<timestamp>\d+) "
- r"(?P<id>\d+) "
- r"(?P<event>is thinking|is eating|is sleeping|died|has taken fork)$",
- line
- )
- if match is None:
- raise error.Format(line, "wrong format")
-
- self._line = line
- self.id = self._parse_ranged_int(match.group("id"), 1, philo_num)
- self.timestamp = self._parse_ranged_int(
- match.group("timestamp"), start_time, current_ms())
-
- self.event = Event.from_string(match.group('event'))
-
- def _parse_ranged_int(self, s, lo, hi):
- try:
- value = int(s)
- if not (lo <= value <= hi):
- raise error.Format(self._line,
- "{} should be between {} - {}".format(s, lo, hi))
- except ValueError:
- raise error.Format(self._line, "{} sould be an integer".format(s))
- return value
-
- def __repr__(self):
- return "{}ms #{} {}".format(self.timestamp, self.id, self.event)
-
-
-class Philo:
- def __init__(
- self,
- id_: int,
- timeout_die: int,
- timeout_eat: int,
- timeout_sleep: int,
- meal_num: int
- ):
- self._logs = []
- self.id = id_
- self._timeout_die = timeout_die
- self._timeout_eat = timeout_eat
- self._timeout_sleep = timeout_sleep
- self._meal_num = meal_num
-
- def add_log(self, log):
- self._logs.append(log)
-
- def check(self):
- if len(self._logs) == 0:
- return
- grouped = [(e, list(g)) for e, g in itertools.groupby(self._logs, (lambda x: x.event))]
- for e, g in grouped:
- if e is Event.EATING:
- if len(g) != self._meal_num:
- raise error.Log(self._logs, "should eat {} times".format(self._meal_num))
- else:
- if len(g) != 1:
- raise error.Log(self._logs, "event {} should occur 1 time".format(Event.to_string(e)))
-
- # events = [e for e, _ in grouped]
- for l1, l2 in zip(self._logs, self._logs[1:]):
- if l2.event is Event.DIED:
- break
- if l1.event is Event.EATING and l2.event is Event.EATING:
- if l2.timestamp - l1.timestamp > self._timeout_eat:
- raise ValueError
- second, timeout = {
- Event.THINKING: (Event.EATING, None),
- Event.EATING: (Event.SLEEPING, self._timeout_eat),
- Event.SLEEPING: (Event.THINKING, self._timeout_sleep)
- }[l1.event]
- if l2.event is not second:
- raise error.Log(self._logs, "invalid switch {} -> {}".format(l1.event, l2.event))
- if timeout is not None and l2.timestamp - l1.timestamp > timeout:
- raise ValueError
-
- last_eat = None
- for log in reversed(self._logs):
- if log.event is Event.EATING:
- last_eat = log
- break
- last = self._logs[-1]
- if last_eat is not None and last_eat is not last:
- if last.timestamp - last_eat.timestamp > self._timeout_die + 10:
- raise error.Log(self._logs, "{} should be dead {} - {} > {}".format(
- self.id, last.timestamp, last_eat.timestamp, self._timeout_die + 10))
-
- def _check_fork_taking(self):
- for l1, l2, l3 in zip(self._logs, self._logs[1:], self._logs[2:]):
- if l1.event is Event.FORK and (l2.event is not Event.FORK or l2.event is not Event.EAT):
- raise ValueError("take fork but no eat")
-
- def _check_meal(self):
- pass
-
- def _check_order(self):
- pass
-
-
- @property
- def last_event(self):
- if len(self._logs) == 0:
- return Event.NONE
- return self._logs[-1].event
-
-
-class Table:
- def __init__(
- self,
- philo_num: int,
- timeout_die: int,
- timeout_eat: int,
- timeout_sleep: int,
- meal_num: int
- ):
- self._philos = [Philo(id_, timeout_die, timeout_eat, timeout_sleep, meal_num)
- for id_ in range(1, philo_num + 1)]
- self._logs = []
- self._philo_num = philo_num
- self.dead = False
-
- def add_log(self, log):
- self._logs.append(log)
- philo = next(p for p in self._philos if p.id == log.id)
- philo.add_log(log)
- if self.dead:
- raise error.Log(self._logs, "should not output after death")
- if log.event is Event.DIED:
- self.dead = True
-
- def check(self):
- if self.dead:
- return
- fork_used = 2 * len([p for p in self._philos if p.last_event == Event.EATING])
- if fork_used > self._philo_num:
- raise error.Log(self._logs, "using nonexistant forks")
- for l1, l2 in zip(self._logs, self._logs[1:]):
- if l1.timestamp > l2.timestamp:
- raise error.Log(self._logs, "timestamp not in ordered")
- for p in self._philos:
- p.check()
diff --git a/src/test/test.py b/src/test/test.py
deleted file mode 100644
index 8a63bd7..0000000
--- a/src/test/test.py
+++ /dev/null
@@ -1,148 +0,0 @@
-# ############################################################################ #
-# #
-# ::: :::::::: #
-# test.py :+: :+: :+: #
-# +:+ +:+ +:+ #
-# By: charles <me@cacharle.xyz> +#+ +:+ +#+ #
-# +#+#+#+#+#+ +#+ #
-# Created: 2020/09/27 11:36:32 by charles #+# #+# #
-# Updated: 2020/09/29 15:15:46 by cacharle ### ########.fr #
-# #
-# ############################################################################ #
-
-import os
-import time
-import subprocess
-
-import config
-import test.philo as philo
-import test.error as error
-from helper import current_ms
-
-
-class Test:
- _tests = []
- _exec_path = None
- _fail_summaries = []
-
- @classmethod
- def run_all(cls, exec_path: str):
- cls._exec_path = exec_path
- for t in cls._tests:
- t.run()
-
- @staticmethod
- def new_error(error_cmd: [str]):
- Test(error_cmd=error_cmd)
-
- def __init__(
- self,
- philo_num: int = None,
- timeout_die: int = None,
- timeout_eat: int = None,
- timeout_sleep: int = None,
- meal_num: int = None,
- error_cmd: [str] = None,
- infinite: bool = False
- ):
- self._philo_num = philo_num
- self._timeout_die = timeout_die
- self._timeout_eat = timeout_eat
- self._timeout_sleep = timeout_sleep
- self._meal_num = meal_num
- self._error_cmd = error_cmd
- self._infinite = infinite
- Test._tests.append(self)
-
- def run(self):
- try:
- self._run_tested()
- except error.Philo as e:
- self._print_fail(e.summary)
- Test._fail_summaries.append(self._argv_str + '\n' + e.full_summary)
- else:
- self._print_pass()
-
- @classmethod
- def write_failed(cls):
- with open(config.RESULT_FILE, "w") as f:
- f.write('\n\n'.join(cls._fail_summaries))
-
- def _run_tested(self):
- process = subprocess.Popen(
- self._argv(),
- stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT
- )
- if self._error_cmd is not None:
- self._check_error(process)
- elif self._infinite:
- try:
- self._check_output(process.stdout, died=False)
- process.wait(timeout=config.INFINITE_WAIT_TIME)
- except subprocess.TimeoutExpired:
- pass
- else:
- raise ShouldBeInfinite
- else:
- self._check_output(process.stdout)
- process.wait(timeout=config.TIMEOUT)
-
- def _check_output(self, stream, died: bool = True):
- start_time = current_ms()
- table = philo.Table(
- self._philo_num, self._timeout_die, self._timeout_eat, self._timeout_sleep,
- 1 if self._meal_num is None else self._meal_num)
- for i, line in enumerate(stream):
- line = line.decode()[:-1]
- table.add_log(philo.Log(line, self._philo_num, start_time))
- table.check()
- if i > 1000:
- break
- if died:
- if not table.dead and self._philo_num != 0:
- raise philo.error.Log(table._logs, "one philosopher should have died")
- else:
- if table.dead:
- raise philo.error.Log(table._logs, "infinite shouldn't die")
-
-
- def _check_error(self, process):
- try:
- out, _ = process.communicate(timeout=config.TIMEOUT_ERROR)
- except subprocess.TimeoutExpired:
- raise error.ShouldFail("no error message")
- if process.returncode == 0:
- raise error.ShouldFail("non zero status code: {}".format(process.returncode))
- if out.decode().count('\n') != 1:
- raise error.ShouldFail("no error message")
-
- def _argv(self, basename=False):
- exec_path = os.path.basename(Test._exec_path) if basename else Test._exec_path
- if self._error_cmd is not None:
- return [exec_path, *self._error_cmd]
- else:
- argv = [
- exec_path,
- str(self._philo_num),
- str(self._timeout_die),
- str(self._timeout_eat),
- str(self._timeout_sleep)
- ]
- if self._meal_num is not None:
- argv.append(str(self._meal_num))
- return argv
-
- @property
- def _argv_str(self):
- return ' '.join(self._argv(basename=True))
-
- RED_CHARS = "\033[31m"
- GREEN_CHARS = "\033[32m"
- CLOSE_CHARS = "\033[0m"
-
- def _print_fail(self, msg):
- print("{}[FAIL] {}: {}{}".format(Test.RED_CHARS, self._argv_str, msg, Test.CLOSE_CHARS))
-
- def _print_pass(self):
- print("{}[PASS] {}{}".format(Test.GREEN_CHARS, self._argv_str, Test.CLOSE_CHARS))
generated by cgit (git 2.34.1) at 2026-05-16 05:10:05 +0000