import sqlite3 import unittest from dependency_injector import containers # Runtime import import os _TOP_DIR = os.path.abspath( os.path.sep.join(( os.path.dirname(__file__), "../", )), ) _SAMPLES_DIR = os.path.abspath( os.path.sep.join(( os.path.dirname(__file__), "../samples/", )), ) import sys sys.path.append(_SAMPLES_DIR) from schemasample.services import UserService, AuthService, PhotoService class TestSchemaSingleContainer(unittest.TestCase): def test(self): container = containers.DynamicContainer() container.from_yaml_schema(f"{_SAMPLES_DIR}/schemasample/container-single.yml") container.config.from_dict({ "database": { "dsn": ":memory:", }, "aws": { "access_key_id": "KEY", "secret_access_key": "SECRET", }, "auth": { "token_ttl": 3600, }, }) # User service user_service1 = container.user_service() user_service2 = container.user_service() self.assertIsInstance(user_service1, UserService) self.assertIsInstance(user_service2, UserService) self.assertIsNot(user_service1, user_service2) self.assertIsInstance(user_service1.db, sqlite3.Connection) self.assertIsInstance(user_service2.db, sqlite3.Connection) self.assertIs(user_service1.db, user_service2.db) # Auth service auth_service1 = container.auth_service() auth_service2 = container.auth_service() self.assertIsInstance(auth_service1, AuthService) self.assertIsInstance(auth_service2, AuthService) self.assertIsNot(auth_service1, auth_service2) self.assertIsInstance(auth_service1.db, sqlite3.Connection) self.assertIsInstance(auth_service2.db, sqlite3.Connection) self.assertIs(auth_service1.db, auth_service2.db) self.assertIs(auth_service1.db, container.database_client()) self.assertIs(auth_service2.db, container.database_client()) self.assertEqual(auth_service1.token_ttl, 3600) self.assertEqual(auth_service2.token_ttl, 3600) # Photo service photo_service1 = container.photo_service() photo_service2 = container.photo_service() self.assertIsInstance(photo_service1, PhotoService) self.assertIsInstance(photo_service2, PhotoService) self.assertIsNot(photo_service1, photo_service2) self.assertIsInstance(photo_service1.db, sqlite3.Connection) self.assertIsInstance(photo_service2.db, sqlite3.Connection) self.assertIs(photo_service1.db, photo_service2.db) self.assertIs(photo_service1.db, container.database_client()) self.assertIs(photo_service2.db, container.database_client()) self.assertIs(photo_service1.s3, photo_service2.s3) self.assertIs(photo_service1.s3, container.s3_client()) self.assertIs(photo_service2.s3, container.s3_client()) class TestSchemaMultipleContainers(unittest.TestCase): def test(self): container = containers.DynamicContainer() container.from_yaml_schema(f"{_SAMPLES_DIR}/schemasample/container-multiple.yml") container.core.config.from_dict({ "database": { "dsn": ":memory:", }, "aws": { "access_key_id": "KEY", "secret_access_key": "SECRET", }, "auth": { "token_ttl": 3600, }, }) # User service user_service1 = container.services.user() user_service2 = container.services.user() self.assertIsInstance(user_service1, UserService) self.assertIsInstance(user_service2, UserService) self.assertIsNot(user_service1, user_service2) self.assertIsInstance(user_service1.db, sqlite3.Connection) self.assertIsInstance(user_service2.db, sqlite3.Connection) self.assertIs(user_service1.db, user_service2.db) # Auth service auth_service1 = container.services.auth() auth_service2 = container.services.auth() self.assertIsInstance(auth_service1, AuthService) self.assertIsInstance(auth_service2, AuthService) self.assertIsNot(auth_service1, auth_service2) self.assertIsInstance(auth_service1.db, sqlite3.Connection) self.assertIsInstance(auth_service2.db, sqlite3.Connection) self.assertIs(auth_service1.db, auth_service2.db) self.assertIs(auth_service1.db, container.gateways.database_client()) self.assertIs(auth_service2.db, container.gateways.database_client()) self.assertEqual(auth_service1.token_ttl, 3600) self.assertEqual(auth_service2.token_ttl, 3600) # Photo service photo_service1 = container.services.photo() photo_service2 = container.services.photo() self.assertIsInstance(photo_service1, PhotoService) self.assertIsInstance(photo_service2, PhotoService) self.assertIsNot(photo_service1, photo_service2) self.assertIsInstance(photo_service1.db, sqlite3.Connection) self.assertIsInstance(photo_service2.db, sqlite3.Connection) self.assertIs(photo_service1.db, photo_service2.db) self.assertIs(photo_service1.db, container.gateways.database_client()) self.assertIs(photo_service2.db, container.gateways.database_client()) self.assertIs(photo_service1.s3, photo_service2.s3) self.assertIs(photo_service1.s3, container.gateways.s3_client()) self.assertIs(photo_service2.s3, container.gateways.s3_client()) class TestSchemaMultipleContainersReordered(unittest.TestCase): def test(self): container = containers.DynamicContainer() container.from_yaml_schema(f"{_SAMPLES_DIR}/schemasample/container-multiple-reordered.yml") container.core.config.from_dict({ "database": { "dsn": ":memory:", }, "aws": { "access_key_id": "KEY", "secret_access_key": "SECRET", }, "auth": { "token_ttl": 3600, }, }) # User service user_service1 = container.services.user() user_service2 = container.services.user() self.assertIsInstance(user_service1, UserService) self.assertIsInstance(user_service2, UserService) self.assertIsNot(user_service1, user_service2) self.assertIsInstance(user_service1.db, sqlite3.Connection) self.assertIsInstance(user_service2.db, sqlite3.Connection) self.assertIs(user_service1.db, user_service2.db) # Auth service auth_service1 = container.services.auth() auth_service2 = container.services.auth() self.assertIsInstance(auth_service1, AuthService) self.assertIsInstance(auth_service2, AuthService) self.assertIsNot(auth_service1, auth_service2) self.assertIsInstance(auth_service1.db, sqlite3.Connection) self.assertIsInstance(auth_service2.db, sqlite3.Connection) self.assertIs(auth_service1.db, auth_service2.db) self.assertIs(auth_service1.db, container.gateways.database_client()) self.assertIs(auth_service2.db, container.gateways.database_client()) self.assertEqual(auth_service1.token_ttl, 3600) self.assertEqual(auth_service2.token_ttl, 3600) # Photo service photo_service1 = container.services.photo() photo_service2 = container.services.photo() self.assertIsInstance(photo_service1, PhotoService) self.assertIsInstance(photo_service2, PhotoService) self.assertIsNot(photo_service1, photo_service2) self.assertIsInstance(photo_service1.db, sqlite3.Connection) self.assertIsInstance(photo_service2.db, sqlite3.Connection) self.assertIs(photo_service1.db, photo_service2.db) self.assertIs(photo_service1.db, container.gateways.database_client()) self.assertIs(photo_service2.db, container.gateways.database_client()) self.assertIs(photo_service1.s3, photo_service2.s3) self.assertIs(photo_service1.s3, container.gateways.s3_client()) self.assertIs(photo_service2.s3, container.gateways.s3_client()) class TestSchemaMultipleContainersWithInlineProviders(unittest.TestCase): def test(self): container = containers.DynamicContainer() container.from_yaml_schema(f"{_SAMPLES_DIR}/schemasample/container-multiple-inline.yml") container.core.config.from_dict({ "database": { "dsn": ":memory:", }, "aws": { "access_key_id": "KEY", "secret_access_key": "SECRET", }, "auth": { "token_ttl": 3600, }, }) # User service user_service1 = container.services.user() user_service2 = container.services.user() self.assertIsInstance(user_service1, UserService) self.assertIsInstance(user_service2, UserService) self.assertIsNot(user_service1, user_service2) self.assertIsInstance(user_service1.db, sqlite3.Connection) self.assertIsInstance(user_service2.db, sqlite3.Connection) self.assertIs(user_service1.db, user_service2.db) # Auth service auth_service1 = container.services.auth() auth_service2 = container.services.auth() self.assertIsInstance(auth_service1, AuthService) self.assertIsInstance(auth_service2, AuthService) self.assertIsNot(auth_service1, auth_service2) self.assertIsInstance(auth_service1.db, sqlite3.Connection) self.assertIsInstance(auth_service2.db, sqlite3.Connection) self.assertIs(auth_service1.db, auth_service2.db) self.assertIs(auth_service1.db, container.gateways.database_client()) self.assertIs(auth_service2.db, container.gateways.database_client()) self.assertEqual(auth_service1.token_ttl, 3600) self.assertEqual(auth_service2.token_ttl, 3600) # Photo service photo_service1 = container.services.photo() photo_service2 = container.services.photo() self.assertIsInstance(photo_service1, PhotoService) self.assertIsInstance(photo_service2, PhotoService) self.assertIsNot(photo_service1, photo_service2) self.assertIsInstance(photo_service1.db, sqlite3.Connection) self.assertIsInstance(photo_service2.db, sqlite3.Connection) self.assertIs(photo_service1.db, photo_service2.db) self.assertIs(photo_service1.db, container.gateways.database_client()) self.assertIs(photo_service2.db, container.gateways.database_client()) self.assertIs(photo_service1.s3, photo_service2.s3) self.assertIs(photo_service1.s3, container.gateways.s3_client()) self.assertIs(photo_service2.s3, container.gateways.s3_client()) class TestSchemaBoto3Session(unittest.TestCase): @unittest.skip("Boto3 tries to connect to the internet") def test(self): container = containers.DynamicContainer() container.from_yaml_schema(f"{_SAMPLES_DIR}/schemasample/container-boto3-session.yml") container.config.from_dict( { "aws_access_key_id": "key", "aws_secret_access_key": "secret", "aws_session_token": "token", "aws_region_name": "us-east-1", }, ) self.assertEqual(container.s3_client().__class__.__name__, "S3") self.assertEqual(container.sqs_client().__class__.__name__, "SQS")