/usr/local/lib/python3.7/unittest/case.py in testPartExecutor
(arguments ▶)
- self : _Outcome
- <unittest.case._Outcome object at 0x7fb0207e3cd0>
- test_case : ImportExportService
- test_cancel_task_export_instance (c2tests.ec2.importexportservice.ImportExportService)
- isTest : bool
- True
- @contextlib.contextmanager
- def testPartExecutor(self, test_case, isTest=False):
- old_success = self.success
- self.success = True
- try:
- yield
- except KeyboardInterrupt:
- raise
- except SkipTest as e:
- self.success = False
- exc_info : NoneType
- None
- old_success : bool
- True
- __unittest : bool
- True
- _subtest_msg_sentinel : object
- <object object at 0x7fb0513c8fd0>
- DIFF_OMITTED : str
-
Diff is %s characters long. Set self.maxDiff to None to see it.
/usr/local/lib/python3.7/unittest/case.py in run
(arguments ▶)
- self : ImportExportService
- test_cancel_task_export_instance (c2tests.ec2.importexportservice.ImportExportService)
- result : DiagnosticsTestResult
- <xmlrunner.extra.diagnosticstestrunner.DiagnosticsTestResult run=1 errors=0 failures=0>
- with outcome.testPartExecutor(self):
- self.setUp()
- if outcome.success:
- outcome.expecting_failure = expecting_failure
- with outcome.testPartExecutor(self, isTest=True):
- testMethod()
- outcome.expecting_failure = False
- with outcome.testPartExecutor(self):
- self.tearDown()
- expecting_failure : bool
- False
- expecting_failure_class : bool
- False
- expecting_failure_method : bool
- False
- orig_result : DiagnosticsTestResult
- <xmlrunner.extra.diagnosticstestrunner.DiagnosticsTestResult run=1 errors=0 failures=0>
- outcome : _Outcome
- <unittest.case._Outcome object at 0x7fb0207e3cd0>
- testMethod : method
- <bound method ImportExportService.test_cancel_task of <c2tests.ec2.importexportservice.ImportExportService testMethod=test_cancel_task_export_instance>>
- __unittest : bool
- True
- _subtest_msg_sentinel : object
- <object object at 0x7fb0513c8fd0>
- DIFF_OMITTED : str
-
Diff is %s characters long. Set self.maxDiff to None to see it.
/root/c2tests/.venv/lib/python3.7/site-packages/parameterized/parameterized.py in standalone_func
(arguments ▶)
- *a : tuple
- (<c2tests.ec2.importexportservice.ImportExportService testMethod=test_cancel_task_export_instance>,)
- @classmethod
- def param_as_standalone_func(cls, p, func, name):
- @wraps(func)
- def standalone_func(*a):
- return func(*(a + p.args), **p.kwargs)
- standalone_func.__name__ = name
- # place_as is used by py.test to determine what source file should be
- # used for this test.
- p : param
- param(*('export_instance',), **{})
- PY2 : bool
- False
- PY3 : bool
- True
- string_types : tuple
- (<class 'str'>,)
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/ec2/importexportservice.py in test_cancel_task
(arguments ▶)
- self : ImportExportService
- test_cancel_task_export_instance (c2tests.ec2.importexportservice.ImportExportService)
- task_type : str
- export_instance
- EXPORT_INSTANCE_TASK_TYPE,
- ])
- def test_cancel_task(self, task_type):
- """Tests task cancelling."""
- task_id = self.create_task_by_type(task_type, self.test_bucket)
- if task_type in IMPORT_TASK_TYPES:
- self.conn.cancel_import_task(task_id)
- else:
- CANCELLED_TASK_STATUS : str
- cancelled
- COMPLETED_TASK_STATUS : str
- completed
- EXPORT_INSTANCE_TASK_TYPE : str
- export_instance
- EXPORT_VOLUME_TASK_TYPE : str
- export_volume
- FAILED_TASK_STATUS : str
- failed
- FINISHED_TASK_STATUSES : set
- {'completed', 'failed', 'cancelled'}
- IMAGES : dict
- {'raw': 'cirros-0.3.4-x86_64-disk.raw', 'vmdk': 'tiny_core.vmdk', 'ova': 'tiny_core.ova', 'invalid_ova': 'empty.ova'}
- IMPORT_IMAGE_TASK_TYPE : str
- import_image
- IMPORT_SNAPSHOT_TASK_TYPE : str
- import_snapshot
- IMPORT_TASK_TYPES : set
- {'import_snapshot', 'import_image'}
- INSTANCE_DEALLOCATION_DELAY : int
- 65
- P2_PROJECT : str
- p2
- WAIT_INSTANCE_STOPPED_TIMEOUT : int
- 180
- WAIT_TASK_FINISHED_TIMEOUT : int
- 540
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/ec2/importexportservice.py in create_task_by_type
(arguments ▶)
- self : ImportExportService
- test_cancel_task_export_instance (c2tests.ec2.importexportservice.ImportExportService)
- task_type : str
- export_instance
- bucket : str
- test-e03aefab-7701-4b5a-8878-995aec9bbb51
- key : NoneType
- None
- disk_format : str
- vmdk
- image_name : str
- tiny_core.vmdk
- **kwargs : dict
- {}
- return self.conn.create_volume_export_task(volume.id, bucket,
- disk_image_format=disk_format, **kwargs).id
- else:
- image_name = image_name or IMAGES[disk_format]
- _, instance = self.run_instance_from_template(image_name=image_name)
- utils.wait_for_instance_states(self.conn, [instance.id],
- [constants.INSTANCE_STATE_RUNNING])
- self.conn.stop_instances([instance.id], True)
- CANCELLED_TASK_STATUS : str
- cancelled
- COMPLETED_TASK_STATUS : str
- completed
- EXPORT_INSTANCE_TASK_TYPE : str
- export_instance
- EXPORT_VOLUME_TASK_TYPE : str
- export_volume
- FAILED_TASK_STATUS : str
- failed
- FINISHED_TASK_STATUSES : set
- {'completed', 'failed', 'cancelled'}
- IMAGES : dict
- {'raw': 'cirros-0.3.4-x86_64-disk.raw', 'vmdk': 'tiny_core.vmdk', 'ova': 'tiny_core.ova', 'invalid_ova': 'empty.ova'}
- IMPORT_IMAGE_TASK_TYPE : str
- import_image
- IMPORT_SNAPSHOT_TASK_TYPE : str
- import_snapshot
- IMPORT_TASK_TYPES : set
- {'import_snapshot', 'import_image'}
- INSTANCE_DEALLOCATION_DELAY : int
- 65
- P2_PROJECT : str
- p2
- WAIT_INSTANCE_STOPPED_TIMEOUT : int
- 180
- WAIT_TASK_FINISHED_TIMEOUT : int
- 540
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/__init__.py in run_instance_from_template
(arguments ▶)
- self : ImportExportService
- test_cancel_task_export_instance (c2tests.ec2.importexportservice.ImportExportService)
- **kwargs : dict
- {'template_id': 'cmi-612EFDED', 'key_name': 's3pT5Ia', 'public_addressing': False}
- # associate eip by default, but some test cases don't need it
- associate_eip = kwargs.pop("associate_eip", True)
- # don't autoassign eip
- kwargs["public_addressing"] = False
- instance = self.run_instance(**kwargs)
- # allocates and associate public address
- if associate_eip:
- address = self.allocate_address()
- associate_eip : bool
- True
- keypair : KeyPair
- KeyPair:s3pT5Ia
- keypair_path : str
- /tmp/tmpymvbvjms.pub
- snapshot : Snapshot
- Snapshot:snap-70F25F43
- gethostbyname : builtin_function_or_method
- <built-in function gethostbyname>
- UNSIGNED : UNSIGNED
- <botocore.UNSIGNED object at 0x7fb043b6ef90>
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/utils.py in wrap
(arguments ▶)
- *args : tuple
- (<c2tests.ec2.importexportservice.ImportExportService testMethod=test_cancel_task_export_instance>,)
- **kwargs : dict
- {'template_id': 'cmi-612EFDED', 'key_name': 's3pT5Ia', 'public_addressing': False}
- @functools.wraps(func)
- def wrap(*args, **kwargs):
- retries = 0
- while retries <= 10:
- try:
- return func(*args, **kwargs)
- except EC2ResponseError as e:
- if e.error_code == "RequestLimitExceeded":
- retries += 1
- time.sleep(0.1 * retries)
- retries : int
- 0
- _DD : str
- /bin/dd
- _QEMU_IMG : str
- /usr/bin/qemu-img
- LOG : RootLogger
- <RootLogger root (DEBUG)>
- md5 : builtin_function_or_method
- <built-in function openssl_md5>
- RULES : list
- [{'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'tcp', 'from_port': '22', 'to_port': '22'}, {'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'icmp', 'from_port': '-1', 'to_port': '-1'}]
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/__init__.py in run_instance
(arguments ▶)
- self : ImportExportService
- test_cancel_task_export_instance (c2tests.ec2.importexportservice.ImportExportService)
- update_instance : bool
- True
- **kwargs : dict
- {'template_id': 'cmi-612EFDED', 'key_name': 's3pT5Ia', 'public_addressing': False}
- return self.run_instances(
- update_instance=update_instance,
- min_count=kwargs.pop("min_count", 1),
- max_count=kwargs.pop("max_count", 1),
- **kwargs,
- )[0]
- def run_instance_micro(self, **kwargs):
- """Helper for micro instance running."""
- gethostbyname : builtin_function_or_method
- <built-in function gethostbyname>
- UNSIGNED : UNSIGNED
- <botocore.UNSIGNED object at 0x7fb043b6ef90>
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/utils.py in wrap
(arguments ▶)
- *args : tuple
- (<c2tests.ec2.importexportservice.ImportExportService testMethod=test_cancel_task_export_instance>,)
- **kwargs : dict
- {'update_instance': True, 'min_count': 1, 'max_count': 1, 'template_id': 'cmi-612EFDED', 'key_name': 's3pT5Ia', 'public_addressing': False}
- @functools.wraps(func)
- def wrap(*args, **kwargs):
- retries = 0
- while retries <= 10:
- try:
- return func(*args, **kwargs)
- except EC2ResponseError as e:
- if e.error_code == "RequestLimitExceeded":
- retries += 1
- time.sleep(0.1 * retries)
- retries : int
- 0
- _DD : str
- /bin/dd
- _QEMU_IMG : str
- /usr/bin/qemu-img
- LOG : RootLogger
- <RootLogger root (DEBUG)>
- md5 : builtin_function_or_method
- <built-in function openssl_md5>
- RULES : list
- [{'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'tcp', 'from_port': '22', 'to_port': '22'}, {'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'icmp', 'from_port': '-1', 'to_port': '-1'}]
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/__init__.py in run_instances
(arguments ▶)
- self : ImportExportService
- test_cancel_task_export_instance (c2tests.ec2.importexportservice.ImportExportService)
- update_instance : bool
- True
- conn : EC2ConnectionLegacy
- <c2tests.profile.EC2ConnectionLegacy object at 0x7fb04217c310>
- **kwargs : dict
- {'min_count': 1, 'max_count': 1, 'key_name': 's3pT5Ia', 'public_addressing': False, 'subnet_id': 'subnet-27809600'}
- update_instance,
- template_id,
- kwargs,
- )
- instances = utils.run_instances(conn, template_id, **kwargs)
- self.log.warning("Instances: %s", common.pprint(instances))
- self.entities[resources.EC2_INSTANCES] += instances
- if update_instance:
- for instance in instances:
- template_id : str
- cmi-612EFDED
- gethostbyname : builtin_function_or_method
- <built-in function gethostbyname>
- UNSIGNED : UNSIGNED
- <botocore.UNSIGNED object at 0x7fb043b6ef90>
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/utils.py in run_instances
(arguments ▶)
- conn : EC2ConnectionLegacy
- <c2tests.profile.EC2ConnectionLegacy object at 0x7fb04217c310>
- template_id : str
- cmi-612EFDED
- reserved : bool
- False
- wait : bool
- True
- **kwargs : dict
- {'min_count': 1, 'max_count': 1, 'key_name': 's3pT5Ia', 'public_addressing': False, 'subnet_id': 'subnet-27809600'}
- [constants.INSTANCE_STATE_DELETED],
- check_for_errors=False)
- except Exception as e:
- LOG.error("Failed to terminate instances %s: %s.", instance_ids, e)
- raise exc
- def ssh(ip, keypath, *remote_args, user="root", **kwargs):
- """Run SSH client"""
- instance_ids : list
- ['i-FC34D320']
- reservation : Reservation
- Reservation:r-FC34D320
- _DD : str
- /bin/dd
- _QEMU_IMG : str
- /usr/bin/qemu-img
- LOG : RootLogger
- <RootLogger root (DEBUG)>
- md5 : builtin_function_or_method
- <built-in function openssl_md5>
- RULES : list
- [{'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'tcp', 'from_port': '22', 'to_port': '22'}, {'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'icmp', 'from_port': '-1', 'to_port': '-1'}]
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/utils.py in run_instances
(arguments ▶)
- conn : EC2ConnectionLegacy
- <c2tests.profile.EC2ConnectionLegacy object at 0x7fb04217c310>
- template_id : str
- cmi-612EFDED
- reserved : bool
- False
- wait : bool
- True
- **kwargs : dict
- {'min_count': 1, 'max_count': 1, 'key_name': 's3pT5Ia', 'public_addressing': False, 'subnet_id': 'subnet-27809600'}
- try:
- reservation = conn.run_instances(template_id, **kwargs)
- if wait:
- wait_for_instance_states(
- conn, [instance.id for instance in reservation.instances],
- [constants.INSTANCE_STATE_RUNNING])
- return [Instance(reservation, instance) for instance in reservation.instances]
- except Exception as exc:
- if reservation is not None:
- instance_ids = [instance.id for instance in reservation.instances]
- instance_ids : list
- ['i-FC34D320']
- reservation : Reservation
- Reservation:r-FC34D320
- _DD : str
- /bin/dd
- _QEMU_IMG : str
- /usr/bin/qemu-img
- LOG : RootLogger
- <RootLogger root (DEBUG)>
- md5 : builtin_function_or_method
- <built-in function openssl_md5>
- RULES : list
- [{'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'tcp', 'from_port': '22', 'to_port': '22'}, {'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'icmp', 'from_port': '-1', 'to_port': '-1'}]
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/utils.py in wait_for_instance_states
(arguments ▶)
- conn : EC2ConnectionLegacy
- <c2tests.profile.EC2ConnectionLegacy object at 0x7fb04217c310>
- instance_ids : list
- ['i-FC34D320']
- states : list
- ['running']
- check_for_errors : bool
- True
- raise Error("Instance {0} has disappeared.",
- set(instance_ids).difference(got_ids).pop())
- return True
- wait(10 * constants.MINUTE_SECONDS, check_state)
- def wait_for_extnet_attached(conn, *extnets):
- """Wait for External Network attached state."""
- aws_error_states : list
- ['error', 'failed', 'crashed']
- aws_states : list
- ['running']
- _DD : str
- /bin/dd
- _QEMU_IMG : str
- /usr/bin/qemu-img
- LOG : RootLogger
- <RootLogger root (DEBUG)>
- md5 : builtin_function_or_method
- <built-in function openssl_md5>
- RULES : list
- [{'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'tcp', 'from_port': '22', 'to_port': '22'}, {'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'icmp', 'from_port': '-1', 'to_port': '-1'}]
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/utils.py in wait
(arguments ▶)
- timeout : int
- 600
- func : function
- <function wait_for_instance_states.<locals>.check_state at 0x7fb03a152d40>
- *args : tuple
- ()
- **kwargs : dict
- {}
- def wait(timeout, func, *args, **kwargs):
- """Fail unless func return success within specified number of seconds"""
- endtime = time.time() + float(timeout)
- while time.time() < endtime:
- result = func(*args, **kwargs)
- if result:
- return result
- time.sleep(min(timeout, 0.1))
- endtime : float
- 1713483272.89324
- result : bool
- False
- _DD : str
- /bin/dd
- _QEMU_IMG : str
- /usr/bin/qemu-img
- LOG : RootLogger
- <RootLogger root (DEBUG)>
- md5 : builtin_function_or_method
- <built-in function openssl_md5>
- RULES : list
- [{'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'tcp', 'from_port': '22', 'to_port': '22'}, {'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'icmp', 'from_port': '-1', 'to_port': '-1'}]
/root/c2tests/.venv/lib/python3.7/site-packages/c2tests/utils.py in check_state
(arguments ▶)
- for reservation in conn.get_all_instances(instance_ids):
- for instance in reservation.instances:
- got_ids.add(instance.id)
- if check_for_errors and instance.state in aws_error_states:
- raise Error("Instance {0} got {1} state.",
- instance.id, instance.state)
- if instance.state not in aws_states:
- return False
- if got_ids != set(instance_ids):
- aws_error_states : list
- ['error', 'failed', 'crashed']
- aws_states : list
- ['running']
- check_for_errors : bool
- True
- conn : EC2ConnectionLegacy
- <c2tests.profile.EC2ConnectionLegacy object at 0x7fb04217c310>
- got_ids : set
- {'i-FC34D320'}
- instance : Instance
- Instance:i-FC34D320
- instance_ids : list
- ['i-FC34D320']
- reservation : Reservation
- Reservation:r-FC34D320
- _DD : str
- /bin/dd
- _QEMU_IMG : str
- /usr/bin/qemu-img
- LOG : RootLogger
- <RootLogger root (DEBUG)>
- md5 : builtin_function_or_method
- <built-in function openssl_md5>
- RULES : list
- [{'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'tcp', 'from_port': '22', 'to_port': '22'}, {'cidr_ip': '0.0.0.0/0', 'ip_protocol': 'icmp', 'from_port': '-1', 'to_port': '-1'}]