Skip to content

Commit

Permalink
adding unit tests
Browse files Browse the repository at this point in the history
  • Loading branch information
Alex Azarh authored and Alex Azarh committed Feb 1, 2017
1 parent c9f1d14 commit 0c0ed14
Show file tree
Hide file tree
Showing 2 changed files with 57 additions and 14 deletions.
44 changes: 30 additions & 14 deletions package/tests/test_connectivity/test_prepare_connectivity.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,17 +20,23 @@ def setUp(self):
self.tag_service = Mock()
self.reservation = Mock()
self.route_table_service = Mock()
self.crypto_service = Mock()

self.prepare_conn = PrepareConnectivityOperation(self.vpc_serv, self.sg_serv, self.key_pair_serv,
self.tag_service, self.route_table_service)
self.tag_service, self.route_table_service,
self.crypto_service)

def test_prepare_conn_command(self):
# Arrange
request = DeployDataHolder({"actions": [
{"actionId": "ba7d54a5-79c3-4b55-84c2-d7d9bdc19356", "actionTarget": None, "customActionAttributes": [
{"attributeName": "Network", "attributeValue": "10.0.0.0/24", "type": "customAttribute"}],
"type": "prepareNetwork"}]})

self.vpc_serv.get_peering_connection_by_reservation_id = Mock(return_value=None)
access_key = Mock()
self.prepare_conn._get_or_create_key_pair = Mock(return_value=access_key)
crypto_dto = Mock()
self.crypto_service.encrypt = Mock(return_value=crypto_dto)

results = self.prepare_conn.prepare_connectivity(ec2_client=self.ec2_client,
ec2_session=self.ec2_session,
Expand All @@ -40,11 +46,18 @@ def test_prepare_conn_command(self):
request=request,
logger=Mock())

self.prepare_conn._get_or_create_key_pair.assert_called_once_with(ec2_session=self.ec2_session,
s3_session=self.s3_session,
bucket=self.aws_dm.key_pairs_location,
reservation_id=self.reservation.reservation_id)
self.crypto_service.encrypt.assert_called_once_with(access_key)
self.assertEqual(request.actions[0].actionId, results[0].actionId)
self.assertEqual(results[0].type, 'PrepareNetwork')
self.assertTrue(results[0].success)
self.assertEqual(results[0].infoMessage, 'PrepareConnectivity finished successfully')
self.assertEqual(results[0].errorMessage, '')
self.assertEqual(results[0].access_key, crypto_dto.encrypted_input)
self.assertEqual(results[0].secret_key, crypto_dto.encrypted_asymmetric_key)

def test_prepare_conn_command_no_management_vpc(self):
request = Mock()
Expand Down Expand Up @@ -80,19 +93,21 @@ def test_prepare_conn_command_fault_res(self):
self.assertIsNotNone(results[0].errorMessage)

def test_create_key_pair(self):
key_pair_service = Mock()
key_pair_service.get_key_for_reservation = Mock(return_value=None)
prepare_conn = PrepareConnectivityOperation(self.vpc_serv, self.sg_serv, key_pair_service, self.tag_service,
self.route_table_service, self.crypto_service)
key_pair = Mock()
key_pair.get_key_for_reservation = Mock(return_value=None)
prepare_conn = PrepareConnectivityOperation(self.vpc_serv, self.sg_serv, key_pair, self.tag_service,
self.route_table_service)
key_pair.create_key_pair = Mock(return_value=Mock())
key_pair_service.create_key_pair = Mock(return_value=key_pair)

prepare_conn._get_or_create_key_pair(self.ec2_session, self.s3_session, 'bucket', 'res_id')
access_key = prepare_conn._get_or_create_key_pair(self.ec2_session, self.s3_session, 'bucket', 'res_id')

self.assertTrue(key_pair.get_key_for_reservation.called_with(self.s3_session, 'bucket', 'res_id'))
self.assertTrue(key_pair.create_key_pair.called_with(self.ec2_session,
self.s3_session,
'bucket',
'res_id'))
self.assertTrue(key_pair_service.get_key_for_reservation.called_with(self.s3_session, 'bucket', 'res_id'))
self.assertTrue(key_pair_service.create_key_pair.called_with(self.ec2_session,
self.s3_session,
'bucket',
'res_id'))
self.assertEquals(access_key, key_pair.key_material)

def test_extract_cidr(self):
action1 = DeployDataHolder({
Expand Down Expand Up @@ -140,7 +155,8 @@ def test_get_or_create_security_group(self):
security_group_service.create_security_group = Mock(return_value=sg)

prepare_conn = PrepareConnectivityOperation(self.vpc_serv, security_group_service, self.key_pair_serv,
self.tag_service, self.route_table_service)
self.tag_service, self.route_table_service,
self.crypto_service)

res = prepare_conn._get_or_create_security_group(self.ec2_session, self.reservation, vpc, management_sg_id)

Expand All @@ -158,7 +174,7 @@ def test_get_or_create_vpc(self):
vpc_service.create_vpc_for_reservation = Mock(return_value=vpc)

prepare_conn = PrepareConnectivityOperation(vpc_service, self.sg_serv, self.key_pair_serv, self.tag_service,
self.route_table_service)
self.route_table_service, self.crypto_service)
cidr = Mock()
prepare_conn._extract_cidr = Mock(return_value=cidr)
res = prepare_conn._get_or_create_vpc(action, self.ec2_session, reservation_id)
Expand Down
27 changes: 27 additions & 0 deletions package/tests/test_domain_services/test_cryptography_service.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
from unittest import TestCase
from mock import Mock, patch

from cloudshell.cp.aws.domain.services.crypto.cryptography import CryptographyService


class TestCryptographyService(TestCase):
def setUp(self):
self.crypto_service = CryptographyService()
self.crypto_service.rsa_service = Mock()

def test_encrypt(self):
# Arrange
plain_text = "password"

encrypted_secret_key = Mock()
self.crypto_service.rsa_service.encrypt = Mock(return_value=encrypted_secret_key)

# Act
crypto_dto = self.crypto_service.encrypt(plain_text)

# Verify
self.crypto_service.rsa_service.encrypt.assert_called_once()
self.assertEquals(crypto_dto.encrypted_asymmetric_key, encrypted_secret_key)
self.assertNotEquals(crypto_dto.encrypted_input, plain_text)
self.assertTrue(crypto_dto.encrypted_input)

0 comments on commit 0c0ed14

Please sign in to comment.