add pytest case
This commit is contained in:
@@ -6,63 +6,74 @@ import typing as t
|
|||||||
|
|
||||||
from app.core import config, security
|
from app.core import config, security
|
||||||
from app.db.session import Base, get_db
|
from app.db.session import Base, get_db
|
||||||
from app.db import models
|
from app.db import models,schemas
|
||||||
from app.main import app
|
from app.main import app
|
||||||
|
|
||||||
|
SQLALCHEMY_DATABASE_URI = f"postgresql+psycopg2://kabAdmin:P%40ssw0rd!@kintonetooldb.postgres.database.azure.com/test"
|
||||||
|
|
||||||
def get_test_db_url() -> str:
|
engine = create_engine(SQLALCHEMY_DATABASE_URI,echo=True)
|
||||||
return f"{config.SQLALCHEMY_DATABASE_URI}"
|
|
||||||
|
|
||||||
|
test_session_maker = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture(scope="function")
|
||||||
def test_db():
|
def test_db():
|
||||||
"""
|
|
||||||
Modify the db session to automatically roll back after each test.
|
|
||||||
This is to avoid tests affecting the database state of other tests.
|
|
||||||
"""
|
|
||||||
# Connect to the test database
|
|
||||||
engine = create_engine(
|
|
||||||
get_test_db_url(),
|
|
||||||
)
|
|
||||||
|
|
||||||
connection = engine.connect()
|
connection = engine.connect()
|
||||||
trans = connection.begin()
|
transaction = connection.begin()
|
||||||
|
test_session = test_session_maker(bind=connection)
|
||||||
# Run a parent transaction that can roll back all changes
|
|
||||||
test_session_maker = sessionmaker(
|
|
||||||
autocommit=False, autoflush=False, bind=engine
|
|
||||||
)
|
|
||||||
test_session = test_session_maker()
|
|
||||||
#test_session.begin_nested()
|
|
||||||
|
|
||||||
# @event.listens_for(test_session, "after_transaction_end")
|
|
||||||
# def restart_savepoint(s, transaction):
|
|
||||||
# if transaction.nested and not transaction._parent.nested:
|
|
||||||
# s.expire_all()
|
|
||||||
# s.begin_nested()
|
|
||||||
|
|
||||||
yield test_session
|
yield test_session
|
||||||
|
|
||||||
# Roll back the parent transaction after the test is complete
|
|
||||||
test_session.close()
|
test_session.close()
|
||||||
trans.rollback()
|
transaction.rollback()
|
||||||
connection.close()
|
connection.close()
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def test_client(test_db):
|
def test_client(test_db):
|
||||||
"""
|
|
||||||
Get a TestClient instance that reads/write to the test database.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def get_test_db():
|
def get_test_db():
|
||||||
|
try:
|
||||||
yield test_db
|
yield test_db
|
||||||
|
finally:
|
||||||
|
test_db.close()
|
||||||
|
|
||||||
app.dependency_overrides[get_db] = get_test_db
|
app.dependency_overrides[get_db] = get_test_db
|
||||||
with TestClient(app) as test_client:
|
with TestClient(app) as test_client:
|
||||||
yield test_client
|
yield test_client
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="function")
|
||||||
|
def test_user(test_db):
|
||||||
|
password ="test"
|
||||||
|
user = models.User(
|
||||||
|
email = "test@test.com",
|
||||||
|
first_name = "test",
|
||||||
|
last_name = "abc",
|
||||||
|
hashed_password = security.get_password_hash(password),
|
||||||
|
is_active = True,
|
||||||
|
is_superuser = False
|
||||||
|
)
|
||||||
|
test_db.add(user)
|
||||||
|
test_db.commit()
|
||||||
|
test_db.refresh(user)
|
||||||
|
dicUser = user.__dict__
|
||||||
|
dicUser["password"] = password
|
||||||
|
return dicUser
|
||||||
|
|
||||||
|
@pytest.fixture(scope="function")
|
||||||
|
def test_admin(test_db):
|
||||||
|
password ="admin"
|
||||||
|
user = models.User(
|
||||||
|
email = "admin@test.com",
|
||||||
|
first_name = "admin",
|
||||||
|
last_name = "abc",
|
||||||
|
hashed_password = security.get_password_hash(password),
|
||||||
|
is_active = True,
|
||||||
|
is_superuser = True
|
||||||
|
)
|
||||||
|
test_db.add(user)
|
||||||
|
test_db.commit()
|
||||||
|
test_db.refresh(user)
|
||||||
|
dicUser = user.__dict__
|
||||||
|
dicUser["password"] = password
|
||||||
|
return dicUser
|
||||||
|
|
||||||
# @pytest.fixture
|
# @pytest.fixture
|
||||||
# def test_password() -> str:
|
# def test_password() -> str:
|
||||||
# return "securepassword"
|
# return "securepassword"
|
||||||
|
|||||||
8
backend/app/tests/test_auth.py
Normal file
8
backend/app/tests/test_auth.py
Normal file
@@ -0,0 +1,8 @@
|
|||||||
|
|
||||||
|
def test_usr_login(test_client,test_user):
|
||||||
|
response = test_client.post("/api/token", data={"username": test_user["email"], "password": test_user["password"]})
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert "access_token" in response.json()
|
||||||
|
assert "token_type" in response.json()
|
||||||
|
assert response.json()["user_name"] == test_user["first_name"]+ " " + test_user["last_name"]
|
||||||
|
|
||||||
22
backend/app/tests/test_user.py
Normal file
22
backend/app/tests/test_user.py
Normal file
@@ -0,0 +1,22 @@
|
|||||||
|
|
||||||
|
def test_users_list(test_client,test_user,test_admin):
|
||||||
|
response = test_client.post("/api/token", data={"username": test_user["email"], "password": test_user["password"]})
|
||||||
|
assert "access_token" in response.json()
|
||||||
|
user_token =response.json()["access_token"]
|
||||||
|
|
||||||
|
response = test_client.post("/api/token", data={"username": test_admin["email"], "password": test_admin["password"]})
|
||||||
|
assert "access_token" in response.json()
|
||||||
|
admin_token =response.json()["access_token"]
|
||||||
|
|
||||||
|
response = test_client.get("/api/v1/users", headers={"Authorization": "Bearer " + user_token})
|
||||||
|
assert response.status_code == 200
|
||||||
|
data = response.json()
|
||||||
|
assert len(data["data"]) == 1
|
||||||
|
|
||||||
|
response = test_client.get("/api/v1/users", headers={"Authorization": "Bearer " + admin_token})
|
||||||
|
assert response.status_code == 200
|
||||||
|
data = response.json()
|
||||||
|
assert len(data["data"]) == 2
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
Reference in New Issue
Block a user