⚝
One Hat Cyber Team
⚝
Your IP:
216.73.216.118
Server IP:
77.37.48.21
Server:
Linux sg-nme-web1518.main-hosting.eu 5.14.0-611.16.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Mon Dec 22 03:40:39 EST 2025 x86_64
Server Software:
LiteSpeed
PHP Version:
8.3.28
Buat File
|
Buat Folder
Eksekusi
Dir :
~
/
opt
/
gsutil
/
gslib
/
vendored
/
boto
/
tests
/
unit
/
swf
/
View File Name :
test_layer2_domain.py
import boto.swf.layer2 from boto.swf.layer2 import Domain, ActivityType, WorkflowType, WorkflowExecution from tests.unit import unittest from mock import Mock class TestDomain(unittest.TestCase): def setUp(self): boto.swf.layer2.Layer1 = Mock() self.domain = Domain(name='test-domain', description='My test domain') self.domain.aws_access_key_id = 'inheritable access key' self.domain.aws_secret_access_key = 'inheritable secret key' self.domain.region = 'test-region' def test_domain_instantiation(self): self.assertEqual('test-domain', self.domain.name) self.assertEqual('My test domain', self.domain.description) def test_domain_list_activities(self): self.domain._swf.list_activity_types.return_value = { 'typeInfos': [{'activityType': {'name': 'DeleteLocalFile', 'version': '1.0'}, 'creationDate': 1332853651.235, 'status': 'REGISTERED'}, {'activityType': {'name': 'DoUpdate', 'version': 'test'}, 'creationDate': 1333463734.528, 'status': 'REGISTERED'}, {'activityType': {'name': 'GrayscaleTransform', 'version': '1.0'}, 'creationDate': 1332853651.18, 'status': 'REGISTERED'}, {'activityType': {'name': 'S3Download', 'version': '1.0'}, 'creationDate': 1332853651.264, 'status': 'REGISTERED'}, {'activityType': {'name': 'S3Upload', 'version': '1.0'}, 'creationDate': 1332853651.314, 'status': 'REGISTERED'}, {'activityType': {'name': 'SepiaTransform', 'version': '1.1'}, 'creationDate': 1333373797.734, 'status': 'REGISTERED'}]} expected_names = ('DeleteLocalFile', 'GrayscaleTransform', 'S3Download', 'S3Upload', 'SepiaTransform', 'DoUpdate') activity_types = self.domain.activities() self.assertEqual(6, len(activity_types)) for activity_type in activity_types: self.assertIsInstance(activity_type, ActivityType) self.assertTrue(activity_type.name in expected_names) self.assertEqual(self.domain.region, activity_type.region) def test_domain_list_workflows(self): self.domain._swf.list_workflow_types.return_value = { 'typeInfos': [{'creationDate': 1332853651.136, 'description': 'Image processing sample workflow type', 'status': 'REGISTERED', 'workflowType': {'name': 'ProcessFile', 'version': '1.0'}}, {'creationDate': 1333551719.89, 'status': 'REGISTERED', 'workflowType': {'name': 'test_workflow_name', 'version': 'v1'}}]} expected_names = ('ProcessFile', 'test_workflow_name') workflow_types = self.domain.workflows() self.assertEqual(2, len(workflow_types)) for workflow_type in workflow_types: self.assertIsInstance(workflow_type, WorkflowType) self.assertTrue(workflow_type.name in expected_names) self.assertEqual(self.domain.aws_access_key_id, workflow_type.aws_access_key_id) self.assertEqual(self.domain.aws_secret_access_key, workflow_type.aws_secret_access_key) self.assertEqual(self.domain.name, workflow_type.domain) self.assertEqual(self.domain.region, workflow_type.region) def test_domain_list_executions(self): self.domain._swf.list_open_workflow_executions.return_value = { 'executionInfos': [{'cancelRequested': False, 'execution': {'runId': '12OeDTyoD27TDaafViz/QIlCHrYzspZmDgj0coIfjm868=', 'workflowId': 'ProcessFile-1.0-1378933928'}, 'executionStatus': 'OPEN', 'startTimestamp': 1378933928.676, 'workflowType': {'name': 'ProcessFile', 'version': '1.0'}}, {'cancelRequested': False, 'execution': {'runId': '12GwBkx4hH6t2yaIh8LYxy5HyCM6HcyhDKePJCg0/ciJk=', 'workflowId': 'ProcessFile-1.0-1378933927'}, 'executionStatus': 'OPEN', 'startTimestamp': 1378933927.919, 'workflowType': {'name': 'ProcessFile', 'version': '1.0'}}, {'cancelRequested': False, 'execution': {'runId': '12oRG3vEWrQ7oYBV+Bqi33Fht+ZRCYTt+tOdn5kLVcwKI=', 'workflowId': 'ProcessFile-1.0-1378933926'}, 'executionStatus': 'OPEN', 'startTimestamp': 1378933927.04, 'workflowType': {'name': 'ProcessFile', 'version': '1.0'}}, {'cancelRequested': False, 'execution': {'runId': '12qrdcpYmad2cjnqJcM4Njm3qrCGvmRFR1wwQEt+a2ako=', 'workflowId': 'ProcessFile-1.0-1378933874'}, 'executionStatus': 'OPEN', 'startTimestamp': 1378933874.956, 'workflowType': {'name': 'ProcessFile', 'version': '1.0'}}]} executions = self.domain.executions() self.assertEqual(4, len(executions)) for wf_execution in executions: self.assertIsInstance(wf_execution, WorkflowExecution) self.assertEqual(self.domain.aws_access_key_id, wf_execution.aws_access_key_id) self.assertEqual(self.domain.aws_secret_access_key, wf_execution.aws_secret_access_key) self.assertEqual(self.domain.name, wf_execution.domain) self.assertEqual(self.domain.region, wf_execution.region) if __name__ == '__main__': unittest.main()