Skip to content

Instantly share code, notes, and snippets.

@Ingco
Created May 13, 2021 13:35
Show Gist options
  • Save Ingco/eccd78e0bc6ecae2957b49902191b8e5 to your computer and use it in GitHub Desktop.
Save Ingco/eccd78e0bc6ecae2957b49902191b8e5 to your computer and use it in GitHub Desktop.
import os
import pytest
from django.apps import apps
from django.conf import settings
from django.core.management import call_command
from django.db import connection
from django.utils import translation
from django_tenants.test.client import TenantClient
from django_tenants.utils import (
get_tenant_model,
get_tenant_domain_model,
get_public_schema_name,
)
@pytest.fixture(scope="session")
def set_lang_en():
translation.activate("en")
def get_basic_actions_control() -> dict:
return {
"main": {
"mycompany": {
"only_one_my_company": {
"action": "redirect",
"properties": {
"args": [
"id"
],
"viewname": "admin:main_mycompany_change"
}
}
}
}
}
def get_basic_disabled_functionality() -> dict:
return {
"main": {
"auth": [
"2fa",
"whitelist_ips"
],
"admin": [
"vacation",
"mycompany__del"
],
"dashboard": [
"current_vacations",
"technicians_salary"
],
"admin_menu": [
"vacation"
],
"inline_admin": [
"vacation"
]
},
"order": {
"pdf": "all",
"admin": [
"ordersetting",
"order__step__add",
"order__step__del",
"order__step__pdf",
"patient__files",
"order__invoice__notes"
]
},
"stock": {
"pdf": "all",
"admin": "all",
"dashboard": "all",
"admin_menu": "all",
"inline_admin": "all"
},
"report": {
"admin_menu": [
"technician",
"all_technician",
"finance",
"stock"
]
},
"finance": {
"admin": "all",
"import": "all",
"reports": "all",
"dashboard": "all",
"admin_menu": "all",
"inline_admin": "all"
},
"invoice": {
"pdf": "all",
"admin": "all",
"import": "all",
"reports": "all",
"dashboard": "all",
"admin_menu": "all",
"inline_admin": "all"
}
}
def get_tenant_schema_name() -> str:
return "test"
def get_full_plan_data() -> dict:
return dict(
title="Full",
disabled_functionality={},
actions_control={},
is_active=True,
is_private=True,
)
@pytest.fixture
def basic_plan_data():
basic_disabled_functionality = get_basic_disabled_functionality()
basic_actions_control = get_basic_actions_control()
return dict(
title="Basic",
disabled_functionality=basic_disabled_functionality,
actions_control=basic_actions_control,
is_active=True,
is_private=True,
)
@pytest.fixture
def full_plan():
plan_model = apps.get_model("saas.Plan")
full_plan_data = get_full_plan_data()
plan_queryset = plan_model.objects.filter(title=full_plan_data["title"])
if plan_queryset.exists():
plan = plan_queryset.first()
yield plan
else:
plan = plan_model(**full_plan_data)
plan.save()
yield plan
# TearDown
plan.delete()
@pytest.fixture
def tenant_schema_data_with_full_plan(full_plan) -> dict:
tenant_schema_name = get_tenant_schema_name()
return dict(
schema_name=tenant_schema_name,
timezone=os.getenv("TENANT_TIMEZONE", "UTC"),
name=tenant_schema_name,
plan_id=full_plan.pk,
)
@pytest.fixture
def tenant_domain_name() -> str:
return os.getenv("TENANT_DOMAIN", "tenant.test.com")
@pytest.fixture
def sync_shared_schema():
call_command(
"migrate_schemas",
schema_name=get_public_schema_name(),
interactive=False,
verbosity=0,
)
def add_allowed_test_domain(tenant_domain_name):
# ALLOWED_HOSTS is a special setting of Django setup_test_environment
# so we can't modify it with helpers
if tenant_domain_name not in settings.ALLOWED_HOSTS:
settings.ALLOWED_HOSTS += [tenant_domain_name]
def remove_allowed_test_domain(tenant_domain_name):
if tenant_domain_name in settings.ALLOWED_HOSTS:
settings.ALLOWED_HOSTS.remove(tenant_domain_name)
@pytest.fixture
def tenant_with_full_plan(tenant_schema_data_with_full_plan: dict):
tenant_model = get_tenant_model()
tenant_queryset = tenant_model.objects.filter(
schema_name=tenant_schema_data_with_full_plan["schema_name"],
)
if tenant_queryset.exists():
tenant = tenant_queryset.first()
yield tenant
else:
tenant = tenant_model(**tenant_schema_data_with_full_plan)
tenant.save()
# Todo: set scope=session
yield tenant
# TearDown
tenant.delete(force_drop=True)
@pytest.fixture
def domain_with_full_plan(tenant_with_full_plan, tenant_domain_name: str):
domain_model = get_tenant_domain_model()
domain_queryset = domain_model.objects.filter(domain=tenant_domain_name)
if domain_queryset.exists():
instance = domain_queryset.first()
return instance
else:
instance = domain_model(
tenant=tenant_with_full_plan,
domain=tenant_domain_name,
)
instance.save()
yield instance
# TearDown
instance.delete()
@pytest.fixture
def setup_connection_with_tenant_with_full_plan(
sync_shared_schema, # apply migrations
tenant_with_full_plan,
domain_with_full_plan,
):
add_allowed_test_domain(tenant_domain_name=domain_with_full_plan.domain)
connection.set_tenant(tenant_with_full_plan)
yield tenant_with_full_plan
# TearDown
connection.set_schema_to_public()
@pytest.fixture
def tenant_api_client_with_full_plan(
set_lang_en,
setup_connection_with_tenant_with_full_plan
):
tenant = setup_connection_with_tenant_with_full_plan
return TenantClient(tenant, CONTENT_TYPE="application/json")
from typing import Optional
import pytest
@pytest.fixture
def invalid_query_params(invoice: int = 0, orders: Optional[str] = None) -> dict:
query_params = {}
if invoice is not None:
query_params["invoice"] = str(invoice)
if orders is not None:
query_params["orders"] = orders
return query_params
@pytest.fixture
def only_invoice_query_params(invoice: int = 1) -> dict:
return {"invoice": str(invoice)}
from typing import Optional
import pytest
from django.urls import reverse
def get_url(invoice: int = 0, orders: Optional[str] = None) -> str:
return reverse("invoice:get_new_invoice_lines")
@pytest.mark.django_db
def test_invalid_without_query_params(
tenant_api_client_with_full_plan,
invalid_query_params: dict,
):
response = tenant_api_client_with_full_plan.get(
get_url(),
invalid_query_params,
)
assert response.status_code == 400
response_data = response.json()
assert response_data.get("status") == "error"
assert "detail" in response_data.keys()
@pytest.mark.django_db
def test_invalid_with_invoice(
tenant_api_client_with_full_plan,
only_invoice_query_params: dict,
):
response = tenant_api_client_with_full_plan.get(
get_url(),
only_invoice_query_params,
)
assert response.status_code == 400
response_data = response.json()
assert response_data.get("status") == "error"
assert "detail" in response_data.keys()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment