pyrestresource/test/test_rest_login.py

from __future__ import annotations
import unittest
from unittest.mock import patch
from os import chdir
from pathlib import Path
from typing import Optional, Annotated
from pydantic import Field
from uuid import UUID, uuid4
from time import time, sleep
from time import time
import json
import uvicorn
import socket
import requests
from contextlib import closing
from multiprocessing import Process
from secrets import token_hex

print(__name__)
print(__package__)

from pydantic import BaseModel

from src.pyrestresource import (
    register_rest_rootpoint,
    RestResourceBase,
    rsrc_verb,
    RestRequestParams_GET,
    RestRequestParams_POST,
    RestRequestParams_Dict_GET,
    RestRequestParams_PUT,
    T_SupportedRESTFields,
    ResourcePlugin_field_default,
    ResourcePlugin_RestResourceBase_default,
)
from src.pyrestresource import ACL_target_user, ACL_target_group, ACL_target_group_Any, ACL_record, ACL_rule
from pprint import pprint

testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())


# to allow mock-ing, all the tested classes are in a function
def init_classes():
    class UserLogin(BaseModel):
        username: str
        secret: str
        token: Optional[str] = None

    class ResourcePlugin_Login(ResourcePlugin_RestResourceBase_default):
        ar_UserLogin: list[UserLogin] = [UserLogin(username="chacha", secret="123456")]

        def handle_resource_get(self, resource: Login, params: RestRequestParams_GET) -> Login:
            print("hook GET")
            print(resource)
            print(params)
            return resource

        def handle_resource_put(self, resource: Login, params: RestRequestParams_GET) -> Login:
            print("hook PUT")

            print(resource.username)
            print(resource.secret)

            for _UserLogin in self.ar_UserLogin:
                if _UserLogin.username == resource.username and _UserLogin.secret == resource.secret:
                    print("user connected")
                    _UserLogin.token = token_hex(16)
                    self.set_resp_cookie("test", _UserLogin.token)
                    print(f"generated token: {_UserLogin.token}")
                    return resource
            print("login NOT found")
            # print(resource)
            # print(resource.username)
            # print(resource.secret)
            # print(params)
            return resource

    class Login(RestResourceBase):
        username: Optional[str] = Field(None)
        secret: Optional[str] = Field(
            None,
            exclude=True,
            ACL=[
                ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.ALLOW),
                ACL_record(verbs=[rsrc_verb.GET, rsrc_verb.DELETE, rsrc_verb.POST], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
            ],
        )

    @register_rest_rootpoint
    class RootApp(RestResourceBase):
        login: Login = Field(default=Login(), plugin=ResourcePlugin_Login)

    # this add the classes to globals to allow using them later on
    # => this is only for uinit-testing purpose and is not needed in real use
    globals()[Login.__name__] = Login
    globals()[RootApp.__name__] = RootApp


def find_free_port():
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
        s.bind(("", 0))
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        hostname = socket.gethostname()
        IPAddr = socket.gethostbyname(hostname)
        return "localhost", s.getsockname()[1]


def launch_server(ip, port):
    print(f"port2={port}")
    init_classes()
    uvicorn.run(f"{__loader__.name}:RootApp", port=port, host="0.0.0.0", log_level="warning", factory=True)


class Test_RestAPI_LOGIN(unittest.TestCase):
    def setUp(self) -> None:
        chdir(testdir_path.parent.resolve())
        init_classes()
        self.testapp = RootApp()

    def test_login(self):
        """
        result = self.testapp.process_request("/login", rsrc_verb.GET)
        print("*****************")
        print(result.get_result())

        result = self.testapp.process_request("/login/username", rsrc_verb.GET)
        print("*****************")
        print(result.get_result())

        # result = self.testapp.process_request("/login/secret", rsrc_verb.GET)
        # print("*****************")
        # print(result.get_result())
        """

        result = self.testapp.process_request("/login", rsrc_verb.PUT, '{"username":"chacha","secret":"123456"}')
        print("*****************")
        print(result.get_result())

        """
        result = self.testapp.process_request("/login", rsrc_verb.GET)
        print("*****************")
        print(result.get_result())

        result = self.testapp.process_request("/login/username", rsrc_verb.GET)
        print("*****************")
        print(result.get_result())

        # result = self.testapp.process_request("/login/secret", rsrc_verb.GET)
        # print("*****************")
        # print(result.get_result())
        """


class Test_RestAPI_LOGIN_Web(unittest.TestCase):
    def setUp(self) -> None:
        chdir(testdir_path.parent.resolve())

    def test_login(self):
        ip, port = find_free_port()
        print(f"ip1={ip}")
        print(f"port1={port}")
        proc = Process(
            target=launch_server,
            args=(
                ip,
                port,
            ),
        )
        proc.start()
        sleep(1)
        s = requests.Session()
        try:
            # Login in

            response = s.put(
                f"http://{ip}:{port}/login",
                json={"username": "chacha", "secret": "123456"},
            )
            print(response)
            print(response.headers)
            self.assertEqual(response.status_code, 201)

            response = s.get(f"http://{ip}:{port}/login")

            response = s.get(f"http://{ip}:{port}/")

        finally:
            proc.terminate()
            s.close()