Fix homecontroller tests

Author: khteh

Pipfile

@@ -28,6 +28,7 @@ quart = "*"
 pytest-flask = "*"
 pytest = "*"
 pytest-asyncio = "*"
+beautifulsoup4 = "*"
 
 [requires]
 python_version = "3.12"

Pipfile.lock

@@ -0,0 +1,8 @@
+# pytest.ini or .pytest.ini
+[pytest]
+minversion = 8.3.2
+addopts = -ra -q
+testpaths =
+    test
+asyncio_mode = auto
+asyncio_default_fixture_loop_scope = function

pytest.ini

@@ -5,7 +5,7 @@
 import quart_flask_patch, json, logging, os
 from quart import Quart, request
 from flask_healthz import HealthError
-from datetime import datetime
+from datetime import datetime, timezone
 from src.controllers.AuthenticationController import auth_api as auth_blueprint
 from src.controllers.UserController import user_api as user_blueprint
 from src.controllers.AuthorController import author_api as author_blueprint

src/app.py

@@ -1,4 +1,5 @@
-import jwt, datetime, logging
+import jwt, logging
+from datetime import datetime, timezone
 from quart import json, Response, session, redirect, url_for, session, abort, current_app
 #from flask_oidc import OpenIDConnect
 from functools import wraps
@@ -19,7 +20,7 @@ def generate_token(user_id):
             raise Exception("Invalid user id!")
         if user_id:
             try:
-                now = datetime.datetime.utcnow()
+                now = datetime.now(timezone.utc)
                 # https://pyjwt.readthedocs.io/en/latest/usage.html#registered-claim-names
                 payload = {
                     "exp": now + datetime.timedelta(hours=1),
@@ -94,7 +95,6 @@ def decorated_auth_required(*args, **kwargs):
 #    @staticmethod
 #    def isAuthenticated():
 #        return oidc.user_loggedin and "user" in session and "token" in session["user"]
-
     @staticmethod
     def require_role(role):
         def decorated_require_role(func):

src/common/Authentication.py

@@ -1,7 +1,7 @@
 import logging
 from quart import request, json, Blueprint, session, render_template, session, redirect, url_for
 from marshmallow import ValidationError
-from datetime import datetime
+from datetime import datetime, timezone
 from base64 import b64decode
 from ..models.UserModel import UserModel, UserSchema
 from ..common.Authentication import Authentication
@@ -11,7 +11,7 @@
 user_schema = UserSchema()
 @auth_api.context_processor
 def inject_now():
-    return {'now': datetime.utcnow()}
+    return {'now': datetime.now(timezone.utc)}
 
 @auth_api.route("/login", methods=["GET", "POST"])
 async def login():
@@ -43,7 +43,7 @@ async def login():
             token = Authentication.generate_token(ser_data.get("id"))
             session["user"] = {"id": ser_data.get("id"), "email": user.email, "token": token}
             #return custom_response({"jwt_token": token}, 200)
-            data["lastlogin"] = datetime.utcnow()
+            data["lastlogin"] = datetime.now(timezone.utc)
             user.update(data)
             logging.info(f"[Auth] User {user.email} logged in")
             if "url" in session and session["url"]:

src/controllers/AuthenticationController.py

@@ -1,15 +1,15 @@
 import re, logging
 from quart import request, Blueprint, session, render_template, flash, redirect, url_for
 from marshmallow import ValidationError
-from datetime import datetime
+from datetime import datetime, timezone
 from ..common.Authentication import Authentication
 from ..common.Response import custom_response
 from ..models.AuthorModel import AuthorModel, AuthorSchema
 author_api = Blueprint("author", __name__)
 author_schema = AuthorSchema()
 @author_api.context_processor
 def inject_now():
-    return {'now': datetime.utcnow()}
+    return {'now': datetime.now(timezone.utc)}
 
 @author_api.route("/index", methods=["GET"])
 async def index():

src/controllers/AuthorController.py

@@ -1,6 +1,6 @@
 import re, logging
 from quart import request, json, Blueprint, session, render_template, flash, redirect, url_for
-from datetime import datetime
+from datetime import datetime, timezone
 from marshmallow import ValidationError
 from ..common.Authentication import Authentication
 from ..common.Response import custom_response
@@ -12,7 +12,7 @@
 author_schema = AuthorSchema()
 @book_api.context_processor
 def inject_now():
-    return {'now': datetime.utcnow()}
+    return {'now': datetime.now(timezone.utc)}
 
 @book_api.route("/index", methods=["GET"])
 async def index():

src/controllers/BookController.py

@@ -1,18 +1,19 @@
 from quart import request, session, Blueprint, flash, render_template, session
 from quart.utils import run_sync
-from datetime import datetime
+from datetime import datetime, timezone
 from array import array
 from ..common.Response import custom_response
 from ..common.Authentication import Authentication
 from ..models.UserModel import UserModel
-import re
+from urllib.parse import urlparse, parse_qs
 fibonacci_api = Blueprint("fibonacci", __name__)
 @fibonacci_api.context_processor
 def inject_now():
-    return {'now': datetime.utcnow()}
+    return {'now': datetime.now(timezone.utc)}
 
 @fibonacci_api.route("/", methods=["GET", "POST"])
 async def fibonacci():
+    print("fibonacci()")
     fibonacci = None
     error = None
     user = None
@@ -26,18 +27,21 @@ async def fibonacci():
         if not user:
             return await render_template("login.html", title="Welcome to Python Flask RESTful API", error="Invalid user!")
     if request.method == "POST":
-        form = await request.form
-        if form['n']:
-            n = form["n"]
-            if n and n.strip():
-                if n.isnumeric():
-                    try:
-                        fibonacci = f"Hello {('there' if not user else user.firstname)}, fibonacci(" + n + "): " + str(await run_sync(fib)(int(n)))
-                        """Renders a greetings page."""
-                    except (Exception) as error:
-                        error = "Exception {0}".format(error)
-                        await flash(f"Fibonacci {n} failed! {error}", "danger")
-                        return await render_template("fibonacci.html", title="Welcome to Python Flask Fibonacci calculator")
+        print("fibonacci() POST")
+        data = await request.get_data()
+        params = parse_qs(data.decode('utf-8'))
+        print(f"data: {data}, params: {params}")
+        if params['n'] and params["n"][0].strip() and params["n"][0].strip().isdigit():
+            n = int(params["n"][0].strip())
+            print(f"fibonacci(): {n}")
+            try:
+                result = await run_sync(fib)(n)
+                print(f"result: {result}")
+                fibonacci = f"Hello {('there' if not user else user.firstname)}, fibonacci({n}): {await run_sync(fib)(n)}"
+            except (Exception) as error:
+                error = "Exception {0}".format(error)
+                await flash(f"Fibonacci {n} failed! {error}", "danger")
+                return await render_template("fibonacci.html", title="Welcome to Python Flask Fibonacci calculator")
         if not fibonacci:
             #error = custom_response({"error": "Please provide an 'N' for the fibonacci number!"}, 400)
             await flash("Please provide a numeric value 'N' for the fibonacci number!", "danger")

src/controllers/FibonacciController.py

@@ -1,12 +1,12 @@
 from quart import Blueprint, render_template, session
-from datetime import datetime
+from datetime import datetime, timezone
 from ..common.Authentication import Authentication
 from ..models.UserModel import UserModel
 import re
 home_api = Blueprint("home", __name__)
 @home_api.context_processor
 def inject_now():
-    return {'now': datetime.utcnow()}
+    return {'now': datetime.now(timezone.utc)}
 
 @home_api.route("/")
 @home_api.route("/index")

src/controllers/HomeController.py

@@ -1,15 +1,15 @@
 import re, logging
 from quart import request, Blueprint, session, render_template, flash, redirect, url_for
 from marshmallow import ValidationError
-from datetime import datetime
+from datetime import datetime, timezone
 from ..models.UserModel import UserModel, UserSchema
 from ..common.Authentication import Authentication
 from ..common.Response import custom_response
 user_api = Blueprint("user", __name__)
 user_schema = UserSchema()
 @user_api.context_processor
 def inject_now():
-    return {'now': datetime.utcnow()}
+    return {'now': datetime.now(timezone.utc)}
 
 @user_api.route("/index", methods=["GET"])
 async def index():

src/controllers/UserController.py

@@ -1,7 +1,7 @@
 from marshmallow import fields, Schema
 from sqlalchemy.sql import func
 from .BookModel import BookSchema
-import datetime
+from datetime import datetime, timezone
 from . import db, bcrypt
 
 class AuthorModel(db.Model):

src/models/AuthorModel.py

@@ -1,7 +1,7 @@
 from . import db
 from marshmallow import fields, Schema
 from sqlalchemy.sql import func
-import datetime
+from datetime import datetime, timezone
 
 class BookModel(db.Model):
     """

src/models/BookModel.py

@@ -1,6 +1,6 @@
 from marshmallow import fields, Schema
 from sqlalchemy.sql import func
-import datetime
+from datetime import datetime, timezone
 from . import db, bcrypt
 from .BookModel import BookSchema
 class UserModel(db.Model):

src/models/UserModel.py

@@ -1,42 +1,23 @@
 import pytest, sys, asyncio
-from hypercorn.config import Config
-from hypercorn.asyncio import serve
 from os.path import dirname, join, abspath
-from src.app import create_app
 #from src.main import app
 from quart_cors import cors
 sys.path.insert(0, abspath(join(dirname(__file__), '../src')))
 from common.Authentication import Authentication
-pytest_plugins = ('pytest_asyncio',)
 
-@pytest.fixture()
-async def app_context():
-    config = Config()
-    config.bind = ["localhost:4433"]
-    config.insecure_bind = ["localhost:8080"]
-    config.worker_class = "asyncio"
-    config.alt_svc_headers = ["h3=\":443\"; ma=3600, h3-29=\":443\"; ma=3600"]
-    config.loglevel = "DEBUG"
-    config.quic_bind = ["localhost:4433"]
-    app = create_app()
-    app = cors(app, allow_credentials=True, allow_origin="https://localhost:4433")
-    asyncio.run(serve(app, config))
-    async with app.app_context() as app_context:
-        yield app_context    
-@pytest.mark.asyncio
-async def test_tokengeneration_pass(app_context):
+def test_tokengeneration_pass():
     """ JWT token generation should pass with valid user input parameter """
     token = Authentication.generate_token("test_user")
     assert type(token) is str
     assert token != ""
-@pytest.mark.asyncio
-async def test_tokengeneration_fail(app_context):
+
+def test_tokengeneration_fail():
     """ JWT token generation should fail without valid user input parameter """
     with pytest.raises(Exception) as e:
         token = Authentication.generate_token("")
     assert "Invalid user id!" in str(e.value)
-@pytest.mark.asyncio
-async def test_tokendecoding_pass(app_context):
+
+def test_tokendecoding_pass():
     """ JWT token decoding should pass with valid user input parameter """
     token = Authentication.generate_token("test_user")
     assert type(token) is str
@@ -49,8 +30,8 @@ async def test_tokendecoding_pass(app_context):
     assert decode["data"]["user_id"] == "test_user"
     expect = dict(data=dict(user_id="test_user"), error=dict())
     assert decode == expect
-@pytest.mark.asyncio
-async def test_tokendecoding_fail(app_context):
+
+def test_tokendecoding_fail():
     """ JWT token decoding should fail without valid user input parameter """
     with pytest.raises(Exception) as e:
         token = Authentication.decode_token("")

test/authentication_test.py

@@ -1,12 +1,32 @@
-import pytest
+import pytest, sys, asyncio
 import os, sys
 from src.main import app
+#from src.app import create_app
+from hypercorn.config import Config
+from hypercorn.asyncio import serve
 from os.path import dirname, join, abspath
 sys.path.insert(0, abspath(join(dirname(__file__), '../src')))
+pytest_plugins = ('pytest_asyncio',)
 
-@pytest.fixture
+@pytest.fixture(scope="function")
 def client():
     return app.test_client()
 
+@pytest.fixture(scope="function")
+async def app_context():
+    config = Config()
+    config.bind = ["localhost:4433"]
+    config.insecure_bind = ["localhost:8080"]
+    config.worker_class = "asyncio"
+    config.alt_svc_headers = ["h3=\":443\"; ma=3600, h3-29=\":443\"; ma=3600"]
+    config.loglevel = "DEBUG"
+    config.quic_bind = ["localhost:4433"]
+    #app = create_app()
+    #app = cors(app, allow_credentials=True, allow_origin="https://localhost:4433")
+    #asyncio.run(serve(app, config))
+    async with app.app_context() as app_context:
+        yield app_context
+
 def pytest_generate_tests(metafunc):
+    """ called once per each test function """
     os.environ['JWT_SECRET_KEY'] = 'pythonflaskrestapipostgres'