add tests for company

Author: surajit

database.py

@@ -5,12 +5,12 @@
 from settings import DATABASE
 from settings import DATABASE_USER
 from settings import DATABASE_PASSWORD
+from settings import DATABASE_DRIVER
+from settings import DATABASE_HOST
 
-SQLALCHEMY_DATABASE_URL = f"postgresql://{DATABASE_USER}:{DATABASE_PASSWORD}@localhost/{DATABASE}"
+SQLALCHEMY_DATABASE_URL = f"{DATABASE_DRIVER}://{DATABASE_USER}:{DATABASE_PASSWORD}@{DATABASE_HOST}/{DATABASE}"
 
-engine = create_engine(
-    SQLALCHEMY_DATABASE_URL
-)
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 Base = declarative_base()

requirements.txt

@@ -13,6 +13,7 @@ Jinja2==3.1.2
 Mako==1.2.0
 MarkupSafe==2.1.1
 mypy-extensions==0.4.3
+mixer==7.2.1
 pytest==5.3.5
 pathspec==0.9.0
 platformdirs==2.5.2
@@ -31,4 +32,5 @@ tomli==2.0.1
 typing_extensions==4.2.0
 urllib3==1.26.9
 zipp==3.8.0
+sqlalchemy-utils==0.38.2
 

routes/v1/contact.py

@@ -20,14 +20,11 @@
 @router.post("/contacts/")
 def create_contact(contact: schema.CreateContact, db: Session = Depends(get_db)):
     db_contact = None
-    hubspot_contact = None
     try:
         org_name = contact.company_name
-        db_org = _company.filter_company_by_name(db, org_name)
-        if db_org:
-            org = db_org.all()[0]
-            db_contact = _contact.create_contact(db, contact, org.id)
-            hubspot_contact = utils.create_contact(data=contact.dict())
+        db_org = _company.filter_company_by_name(db, org_name).all()[0]
+        db_contact = _contact.create_contact(db, contact, db_org.id)
+        hubspot_contact = utils.create_contact(data=contact.dict())
     except (utils.ContactException, ContactExistException) as exc:
         if isinstance(exc, utils.ContactException) and db_contact:
             _contact.delete_contact(db, db_contact.id)

settings.py

@@ -3,16 +3,17 @@
 from decouple import config
 
 DATABASE_HOST = config("DATABASE_HOST")
-DATABASE = config('DATABASE_NAME')
-DATABASE_USER = config('DATABASE_USER')
-DATABASE_PASSWORD = config('DATABASE_PASSWORD')
+DATABASE = config("DATABASE_NAME")
+DATABASE_USER = config("DATABASE_USER")
+DATABASE_PASSWORD = config("DATABASE_PASSWORD")
 DATABASE_DRIVER = config("DATABASE_DRIVER")
+TEST_DATABASE_NAME = config("TEST_DATABASE_NAME")
 
 STRIPE_PUBLISHABLE_KEY = config("STRIPE_PUBLISHABLE_KEY")
 STRIPE_SECRET_KEY = config("STRIPE_SECRET_KEY")
 
 HUBSPOT_API_KEY = config("HUBSPOT_API_KEY")
 
-SENDGRID_API_KEY = config('SENDGRID_API_KEY')
-WELCOME_MESSAGE_TEMPLATE_ID = config('WELCOME_MESSAGE_TEMPLATE_ID')
-PAYMENT_CONFIRMATION_TEMPLATE_ID = config('WELCOME_MESSAGE_TEMPLATE_ID')
+SENDGRID_API_KEY = config("SENDGRID_API_KEY")
+WELCOME_MESSAGE_TEMPLATE_ID = config("WELCOME_MESSAGE_TEMPLATE_ID")
+PAYMENT_CONFIRMATION_TEMPLATE_ID = config("WELCOME_MESSAGE_TEMPLATE_ID")

tests/conftest.py

@@ -0,0 +1,49 @@
+import pytest
+
+from sqlalchemy import create_engine
+from sqlalchemy_utils import database_exists
+from sqlalchemy_utils import create_database
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+import schemas.schema
+from crud.company import create_company
+from tests.test_database import SQLALCHEMY_DATABASE_URL
+
+from dependencies.dependencies import get_db
+from app import app
+from database import Base
+
+
+@pytest.fixture(scope="session")
+def db_engine():
+    engine = create_engine(SQLALCHEMY_DATABASE_URL)
+    if not database_exists:
+        create_database(engine.url)
+
+    Base.metadata.create_all(bind=engine)
+    yield engine
+
+
+@pytest.fixture(scope="function")
+def db(db_engine):
+    connection = db_engine.connect()
+    # begin a non-ORM transaction
+    connection.begin()
+    # bind an individual Session to the connection
+    db = Session(bind=connection)
+    yield db
+    db.rollback()
+    connection.close()
+
+
+@pytest.fixture(scope="function")
+def client(db):
+    app.dependency_overrides[get_db] = lambda: db
+    with TestClient(app) as c:
+        yield c
+
+
+@pytest.fixture
+def organization(db):
+    create_company(db, schemas.schema.CreateCompany(name="Test org", org_id=12345))

tests/routes/test_contact.py

@@ -0,0 +1,45 @@
+from unittest import mock
+
+
+@mock.patch("routes.v1.contact.utils.get_contact_by_email")
+def test_get_contact_by_email(mock_get_contact_by_email, client):
+    mock_get_contact_by_email.return_value = {
+        "id": "1234",
+        "email": "[email protected]",
+        "firstname": "test",
+        "lastname": "user",
+    }
+    response = client.get("/api/v1/contact/[email protected]")
+    assert response.status_code == 200
+    assert response.json() == {
+        "id": "1234",
+        "email": "[email protected]",
+        "firstname": "test",
+        "lastname": "user",
+    }
+
+
+@mock.patch("routes.v1.contact.utils.create_contact")
+@mock.patch("routes.v1.contact._contact.create_contact")
+def test_create_contact(
+    mock_db_create_contact, mock_hubspot_create_contact, organization, client
+):
+    mock_db_create_contact.return_value = {
+        "first_name": "raaj",
+        "last_name": "das",
+        "email": "[email protected]",
+        "phone": "+254720323309",
+    }
+    mock_hubspot_create_contact.return_value = {"contact_id": 12356}
+    response = client.post(
+        "/api/v1/contacts/",
+        json={
+            "first_name": "raaj",
+            "last_name": "das",
+            "email": "[email protected]",
+            "phone": "+255720323309",
+            "company_name": "Test org",
+        },
+    )
+    assert response.status_code == 200
+    assert response.json() == {"contact_id": 12356}

tests/test_database.py

@@ -0,0 +1,26 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from dependencies.dependencies import get_db
+from app import app
+from settings import DATABASE_DRIVER
+from settings import DATABASE_USER
+from settings import DATABASE_PASSWORD
+from settings import DATABASE_HOST
+from settings import TEST_DATABASE_NAME
+
+SQLALCHEMY_DATABASE_URL = f"{DATABASE_DRIVER}://{DATABASE_USER}:{DATABASE_PASSWORD}@{DATABASE_HOST}/{TEST_DATABASE_NAME}"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+def override_get_db():
+    try:
+        db = TestingSessionLocal()
+        yield db
+    finally:
+        db.close()
+
+
+app.dependency_overrides[get_db] = override_get_db