Add unittests (#5)

* add unittest for secret

* fix codecov for all branch

* add unittest for secret

* fix test class name

* fix test_secrete_manager name

* add unittest for resource
add unittest for exception

* fix exceptions.py for mypy

.travis.yml

@@ -18,6 +18,7 @@ script:
 - flake8 --max-line-length 119 local_data_api
 - mypy --ignore-missing-imports local_data_api
 - nosetests -v --with-coverage tests
+- codecov
 - cp README.md docs/index.md && mkdocs build --verbose --clean --strict
 
 deploy:
@@ -30,7 +31,4 @@ deploy:
  github_token: $GITHUB_TOKEN
  local_dir: site
  on:
- branch: master
-
-after_success:
-- codecov
+ branch: master

local_data_api/exceptions.py

@@ -1,9 +1,10 @@
+from abc import ABC
 from typing import Optional, Dict
 
 from fastapi import HTTPException
 
 
-class DataAPIException(HTTPException):
+class DataAPIException(HTTPException, ABC):
  STATUS_CODE: int
 
  def __init__(self, message: str, headers: Optional[Dict] = None):
@@ -24,6 +25,9 @@ class ForbiddenException(DataAPIException):
 
 
 class InternalServerErrorException(DataAPIException):
+ def __init__(self, message: Optional[str] = None):
+ super().__init__(message or 'InternalServerError')
+
  STATUS_CODE = 500
 
 

local_data_api/resources/resource.py

@@ -32,7 +32,7 @@ def set_session(transaction_id: str, session: Session):
  SESSION_POOL[transaction_id] = session
 
 
-def delete_session(transaction_id: Optional[str]) -> None:
+def delete_session(transaction_id: Optional[str] = None) -> None:
  if transaction_id:
  del SESSION_POOL[transaction_id]
 

local_data_api/secret_manager.py

@@ -20,7 +20,8 @@ def create_secret_arn(region_name: str = 'us-east-1', account: str = '1234567890
  return f'arn:aws:secretsmanager:{region_name}:{account}:secret:local-data-api{sha1().hexdigest()}'
 
 
-def register_secret(user_name: Optional[str], password: Optional[str], secret_arn: Optional[str]) -> str:
+def register_secret(user_name: Optional[str] = None, password: Optional[str] = None,
+ secret_arn: Optional[str] = None) -> str:
  if not secret_arn:
  secret_arn = create_secret_arn()
 

tests/test_exceptions.py

@@ -0,0 +1,11 @@
+
+from unittest import TestCase
+
+from local_data_api.exceptions import DataAPIException
+
+
+class TestDataAPIException(TestCase):
+ def test_code(self) -> None:
+ class DummyError(DataAPIException):
+ STATUS_CODE = 500
+ self.assertEqual(DummyError('test').code, 'DummyError')

tests/test_resource/test_resource.py

@@ -3,12 +3,13 @@
 from unittest import TestCase, mock
 from unittest.mock import Mock
 
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import sessionmaker, Session
 
-from local_data_api.exceptions import BadRequestException
+from local_data_api.exceptions import BadRequestException, InternalServerErrorException
 from local_data_api.resources import MySQL
 from local_data_api.resources.sqlite import SQLite
-from local_data_api.resources.resource import Resource, get_resource, register_resource, RESOURCE_METAS, ResourceMeta
+from local_data_api.resources.resource import get_resource, register_resource, RESOURCE_METAS, ResourceMeta, \
+ SESSION_POOL, set_session, get_session, delete_session, get_resource_class, create_resource_arn
 
 DATABASE_SETTINGS: Dict[str, Dict[str, Union[str, int]]] = {
  'SQLite': {
@@ -20,9 +21,10 @@
 }
 
 
-class DataBaseFunction(TestCase):
+class TestResourceFunction(TestCase):
  def setUp(self) -> None:
  RESOURCE_METAS.clear()
+ SESSION_POOL.clear()
 
  def test_register_resource(self) -> None:
  resource_arn: str = 'dummy_resource_arn'
@@ -35,6 +37,7 @@ def test_register_resource(self) -> None:
  self.assertEqual(resource_meta.password, 'pw')
 
  def test_get_resource(self) -> None:
+
  resource_arn: str = 'dummy_resource_arn'
 
  engine = MySQL.create_engine('localhost', 3306, 'test', 'pw', {})
@@ -51,6 +54,64 @@ def test_get_resource(self) -> None:
  mock_get_secret.return_value = secret
  resource = get_resource(resource_arn, 'dummy')
  self.assertIsInstance(resource, SQLite)
+ with mock.patch('local_data_api.resources.resource.get_session') as mock_get_session:
+ new_session = Session()
+ mock_get_session.return_value = new_session
+ resource = get_resource(resource_arn, 'dummy', 'transaction')
+ self.assertEqual(resource.session, new_session)
+
+ def test_get_resource_exception(self) -> None:
+
+ resource_arn: str = 'dummy_resource_arn'
+
+ engine = MySQL.create_engine('localhost', 3306, 'test', 'pw', {})
+
+ session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+ RESOURCE_METAS[resource_arn] = ResourceMeta(SQLite, session, 'localhost', 3306, 'test', 'pw')
+
+ with self.assertRaises(BadRequestException):
+ get_resource('invalid', 'dummy')
+
+ with self.assertRaises(InternalServerErrorException):
+ SESSION_POOL['dummy'] = Session()
+ get_resource('invalid', 'dummy', 'dummy')
+ del SESSION_POOL['dummy']
+
+ with mock.patch('local_data_api.resources.resource.get_secret') as mock_get_secret:
+ with self.assertRaises(BadRequestException):
+ mock_get_secret.side_effect=BadRequestException('error')
+ get_resource(resource_arn, 'dummy')
+
+ with self.assertRaises(InternalServerErrorException):
+ mock_get_secret.side_effect = BadRequestException('error')
+ SESSION_POOL['dummy'] = Session()
+ get_resource(resource_arn, 'dummy', 'dummy')
+
+ mock_get_secret.side_effect = None
+ secret = Mock()
+ secret.user_name = 'invalid'
+ secret.password = 'pw'
+
+ mock_get_secret.return_value = secret
+ with self.assertRaises(BadRequestException):
+ get_resource(resource_arn, 'dummy')
+
+ secret = Mock()
+ secret.user_name = 'test'
+ secret.password = 'invalid'
+
+ mock_get_secret.return_value = secret
+ with self.assertRaises(BadRequestException):
+ get_resource(resource_arn, 'dummy')
+
+ def test_get_resource_class_exception(self) -> None:
+ with self.assertRaises(Exception):
+ get_resource_class('invalid_engine')
+
+ def test_create_resource_arn(self):
+ self.assertEqual(create_resource_arn('us-east-1', '123456789012')[:57],
+ 'arn:aws:rds:us-east-1:123456789012:cluster:local-data-api')
 
  def test_get_database_invalid_resource_arn(self) -> None:
  resource_arn: str = 'dummy_resource_arn'
@@ -63,3 +124,32 @@ def test_get_database_invalid_resource_arn(self) -> None:
 
  with self.assertRaises(BadRequestException):
  get_resource('invalid_arn', 'secret_arn')
+
+
+class TestSessionPool(TestCase):
+ def setUp(self) -> None:
+ SESSION_POOL.clear()
+
+ def test_set_session(self) -> None:
+ session: Session = Session()
+ set_session('abc', session)
+ self.assertEqual(SESSION_POOL['abc'], session)
+
+ def test_get_session(self) -> None:
+ session: Session = Session()
+ SESSION_POOL['abc'] = session
+ self.assertEqual(get_session('abc'), session)
+
+ def test_get_session_notfound(self) -> None:
+ with self.assertRaises(BadRequestException):
+ get_session('abc')
+
+ def test_delete_session(self) -> None:
+ session: Session = Session()
+ SESSION_POOL['abc'] = session
+ delete_session()
+ self.assertEqual(SESSION_POOL['abc'], session)
+
+ delete_session('abc')
+ self.assertTrue('abc' not in SESSION_POOL)
+

tests/test_secret_manager.py

@@ -0,0 +1,54 @@
+
+from unittest import TestCase
+
+from local_data_api.exceptions import BadRequestException
+from local_data_api.secret_manager import register_secret, get_secret, create_secret_arn, SECRETS, Secret
+
+
+class TestSecretManagerFunction(TestCase):
+ def setUp(self) -> None:
+ SECRETS.clear()
+
+ def test_create_secret_arn(self) -> None:
+ self.assertEqual(create_secret_arn()[:67], 'arn:aws:secretsmanager:us-east-1:123456789012:secret:local-data-api')
+ self.assertEqual(create_secret_arn(region_name='ap-northeast-1', account='000000000000')[:72],
+ 'arn:aws:secretsmanager:ap-northeast-1:000000000000:secret:local-data-api')
+
+ def test_register_secret(self) -> None:
+
+ register_secret(user_name='root', password='example',
+ secret_arn='arn:aws:secretsmanager:ap-northeast-1:000000000000:secret:local-data-api')
+
+ secret: Secret = SECRETS['arn:aws:secretsmanager:ap-northeast-1:000000000000:secret:local-data-api']
+ self.assertEqual(secret, Secret('root', 'example'))
+
+ def test_register_secret_no_arn(self) -> None:
+
+ secret_arn: str = register_secret('root', 'example')
+
+ secret: Secret = SECRETS[secret_arn]
+ self.assertEqual(secret, Secret('root', 'example'))
+
+ def test_get_secret(self) -> None:
+ secret_arn = 'arn:aws:secretsmanager:ap-northeast-1:000000000000:secret:local-data-api'
+
+ expected_secret: Secret = Secret('root', 'example')
+ SECRETS[secret_arn] = expected_secret
+
+ secret: Secret = get_secret(secret_arn)
+
+ self.assertEqual(secret, expected_secret)
+
+ def test_get_secret_notfound_secret(self) -> None:
+ secret_arn = 'arn:aws:secretsmanager:ap-northeast-1:000000000000:secret:local-data-api'
+
+ exception = BadRequestException(f'Error fetching secret {secret_arn} : Secrets Manager can’t find the specified '
+ f'secret. (Service: AWSSecretsManager; Status Code: 400; Error Code: '
+ f'ResourceNotFoundException; Request ID: 00000000-1111-2222-3333-44444444444)')
+
+ with self.assertRaises(BadRequestException, ):
+ try:
+ get_secret(secret_arn)
+ except BadRequestException as e:
+ self.assertEqual(e.message, exception.message)
+ raise