srab/autotest/main.py

#!/usr/bin/env python3
import http.client
import importlib
import json
import os
import socket
import sqlite3
import subprocess
import sys
import time
import argparse
from pathlib import Path
import traceback

ROOT = Path(__file__).resolve().parents[1]
DB_PATH = ROOT / "var" / "srab.db"
SERVER_PATH = os.environ.get("SERVER_PATH", ROOT / "build" / "target.exe")
HOST = "127.0.0.1"
PORT = 1337

if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

K_FIRST_NAME = "имя"
K_LAST_NAME = "фамилия"
K_MIDDLE_NAME = "отчество"
K_EDUCATION = "образование"
K_PASSWORD = "пароль"
K_CONFIRM_PASSWORD = "повтор пароля"
K_NEW_PASSWORD = "новый пароль"
K_SNILS = "снилс"
K_PASSPORT = "паспорт"
K_CLASS_NUMBER = "номер"
K_CLASS_LETTER = "буква"
K_CLASSES = "классы"
K_IDENTIFIER = "идентификатор"
K_CREATOR = "создатель"
K_USERNAME = "имя пользователя"

MESSAGE_PASSWORD_MISMATCH = "Пароли не совпадают."
MESSAGE_LOGIN_OK = "Рады видеть вас снова :3"

CASE_PACKAGE = "autotest.cases"
CASE_MODULES = [
    "case_01_users_create_teacher",
    "case_02_users_create_mismatch",
    "case_03_users_login_success",
    "case_04_users_login_failure",
    "case_05_users_change_password",
    "case_06_users_login_with_new_password",
    "case_07_teachers_auth_required",
    "case_08_teachers_create_class",
    "case_09_teachers_list_classes",
    "case_10_teachers_create_extra_class",
    "case_11_teachers_delete_extra_class",
    "case_12_students_create_student",
    "case_13_students_create_second_student",
    "case_14_teachers_add_student_to_class",
    "case_15_students_list",
    "case_16_students_get",
    "case_17_students_delete",
    "case_18_lessons_teacher_add",
    "case_19_lessons_list",
    "case_20_lessons_teacher_delete",
    "case_21_teachers_remove_student_from_class",
]


class TestContext:
    def __init__(self) -> None:
        self.teacher_first = "Alice"
        self.teacher_last = "Smith"
        self.teacher_middle = "Ann"
        self.teacher_education = "Math"
        self.teacher_password_initial = "Secret42"
        self.teacher_password_new = "Secret99"
        self.teacher_password = self.teacher_password_initial
        self.teacher_username = f"{self.teacher_first}.{self.teacher_last}"

        self.class_id = None
        self.class_two_id = None
        self.student_one_id = None
        self.student_two_id = None
        self.teacher_id = None
        self.teacher_user_id = None
        self.lesson_first_id = None
        self.lesson_first_date = None
        self.lesson_first_title = None
        self.lesson_second_id = None
        self.lesson_second_date = None
        self.lesson_second_title = None
        self.student_one_username = None
        self.student_one_password = None
        self.student_two_username = None
        self.student_two_password = None

        self.K_FIRST_NAME = K_FIRST_NAME
        self.K_LAST_NAME = K_LAST_NAME
        self.K_MIDDLE_NAME = K_MIDDLE_NAME
        self.K_EDUCATION = K_EDUCATION
        self.K_PASSWORD = K_PASSWORD
        self.K_CONFIRM_PASSWORD = K_CONFIRM_PASSWORD
        self.K_NEW_PASSWORD = K_NEW_PASSWORD
        self.K_SNILS = K_SNILS
        self.K_PASSPORT = K_PASSPORT
        self.K_CLASS_NUMBER = K_CLASS_NUMBER
        self.K_CLASS_LETTER = K_CLASS_LETTER
        self.K_CLASSES = K_CLASSES
        self.K_IDENTIFIER = K_IDENTIFIER
        self.K_CREATOR = K_CREATOR
        self.K_USERNAME = K_USERNAME
        self.MESSAGE_PASSWORD_MISMATCH = MESSAGE_PASSWORD_MISMATCH
        self.MESSAGE_LOGIN_OK = MESSAGE_LOGIN_OK

    def expect(self, condition: bool, message: str) -> None:
        expect(condition, message)

    def send_request(self, method: str, path: str, *, body=None, headers=None):
        return send_request(method, path, body=body, headers=headers)

    def query_single_value(self, sql: str, params) -> int:
        return query_single_value(sql, params)

    def make_auth(self, username: str, password: str) -> dict:
        return make_auth(username, password)


def remove_database() -> None:
    if DB_PATH.exists():
        DB_PATH.unlink()


def start_server(silent) -> subprocess.Popen:
    out = subprocess.DEVNULL if silent else None

    return subprocess.Popen(
        [str(SERVER_PATH)],
        cwd=str(ROOT),
        stdout=out,
        stderr=out,
    )


def wait_for_server(timeout: float = 10.0) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with socket.create_connection((HOST, PORT), timeout=0.25):
                return
        except OSError:
            time.sleep(0.05)
    raise RuntimeError("server did not accept connections")


def stop_server(proc: subprocess.Popen) -> None:
    if proc.poll() is not None:
        return
    proc.terminate()
    try:
        proc.wait(timeout=2)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait(timeout=2)


def expect(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def make_auth(username: str, password: str) -> dict:
    return {"Authorization": f"Basic {username} {password}"}


def send_request(method: str, path: str, *, body=None, headers=None):
    conn = http.client.HTTPConnection(HOST, PORT, timeout=5)
    try:
        data = None
        hdrs = {} if headers is None else dict(headers)
        if body is not None:
            if isinstance(body, (dict, list)):
                data = json.dumps(body, ensure_ascii=False).encode("utf-8")
                hdrs.setdefault(
                    "Content-Type",
                    "application/json; charset=utf-8",
                )
            elif isinstance(body, str):
                data = body.encode("utf-8")
            else:
                data = body
        conn.request(method, path, body=data, headers=hdrs)
        response = conn.getresponse()
        payload = response.read()
        text = payload.decode("utf-8") if payload else ""
        return response.status, text, dict(response.getheaders())
    finally:
        conn.close()


def query_single_value(sql: str, params) -> int:
    with sqlite3.connect(DB_PATH) as conn:
        conn.row_factory = sqlite3.Row
        row = conn.execute(sql, params).fetchone()
        expect(row is not None, f"query returned no rows: {sql!r} {params!r}")
        return int(row["id"])


def load_cases():
    modules = []
    for name in CASE_MODULES:
        module = importlib.import_module(f"{CASE_PACKAGE}.{name}")
        modules.append(module)
    return modules


def run_tests() -> None:
    context = TestContext()

    for module in load_cases():
        step_name = getattr(module, "TEST_NAME", module.__name__)
        print(f"[step] {step_name}")

        try:
            module.run(context)
        except Exception:
            print(f"Test '{step_name}' run failed")
            print(traceback.format_exc())
            sys.exit(1)

    print("[OK] All steps passed")


def main() -> None:
    parser = argparse.ArgumentParser(description="SRAB automatic test suit.")
    parser.add_argument(
        "-s",
        "--suppress",
        default=False,
        action="store_true",
        help="Suppress server output"
    )
    args = parser.parse_args()

    remove_database()
    server = start_server(args.suppress)

    try:
        wait_for_server()
        run_tests()
    finally:
        stop_server(server)


if __name__ == "__main__":
    main()