/usr/share/pyshared/txaws/testing/ec2.py is in python-txaws 0.2.3-1ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | # Copyright (C) 2009 Canonical Ltd
# Copyright (C) 2009 Duncan McGreggor <oubiwann@adytum.us>
# Licenced under the txaws licence available at /LICENSE in the txaws source.
from datetime import datetime
from twisted.internet.defer import succeed, fail
from twisted.python.failure import Failure
from twisted.web.error import Error
from txaws.ec2.model import Keypair, SecurityGroup
class FakeEC2Client(object):
def __init__(self, creds, endpoint, instances=None, keypairs=None,
volumes=None, key_material="", security_groups=None,
snapshots=None, addresses=None, availability_zones=None,
query_factory=None, parser=None):
self.creds = creds
self.endpoint = endpoint
self.query_factory = query_factory
self.parser = parser
self.instances = instances or []
self.keypairs = keypairs or []
self.keypairs_deleted = []
self.volumes = volumes or []
self.volumes_deleted = []
self.key_material = key_material
self.security_groups = security_groups or []
self.security_groups_deleted = []
self.snapshots = snapshots or []
self.snapshots_deleted = []
self.addresses = addresses or []
self.availability_zones = availability_zones or []
def describe_instances(self, *instances):
return succeed(self.instances)
def run_instances(self, image_id, min_count, max_count,
security_groups=None, key_name=None, instance_type=None,
user_data=None, availability_zone=None, kernel_id=None,
ramdisk_id=None):
return succeed(self.instances)
def terminate_instances(self, *instance_ids):
result = [(instance.instance_id, instance.instance_state,
u"shutting-down") for instance in self.instances]
return succeed(result)
def describe_keypairs(self):
return succeed(self.keypairs)
def create_keypair(self, name):
keypair = Keypair(name, "fingerprint", self.key_material)
return succeed(keypair)
def delete_keypair(self, name):
self.keypairs_deleted.append(name)
return succeed(True)
def describe_security_groups(self, names=None):
return succeed(self.security_groups)
def create_security_group(self, name, description):
self.security_groups.append(SecurityGroup(name, description))
return succeed(True)
def delete_security_group(self, name):
self.security_groups_deleted.append(name)
return succeed(True)
def describe_volumes(self, *volume_ids):
return succeed(self.volumes)
def create_volume(self, availability_zone, size=None, snapshot_id=None):
return succeed(self.volumes[0])
def attach_volume(self, volume_id, instance_id, device):
return succeed({"status": u"attaching",
"attach_time": datetime(2007, 6, 6, 11, 10, 00)})
def delete_volume(self, volume_id):
self.volumes_deleted.append(volume_id)
return succeed(True)
def describe_snapshots(self, *snapshot_ids):
return succeed(self.snapshots)
def create_snapshot(self, volume_id):
return succeed(self.snapshots[0])
def delete_snapshot(self, volume_id):
self.snapshots_deleted.append(volume_id)
return succeed(True)
def authorize_group_permission(self, group_name, source_group_name,
source_group_owner_id):
return succeed(True)
def revoke_group_permission(self, group_name, source_group_name,
source_group_owner_id):
return succeed(True)
def authorize_ip_permission(self, group_name, protocol, from_port, to_port,
cidr_ip):
return succeed(True)
def revoke_ip_permission(self, group_name, protocol, from_port, to_port,
cidr_ip):
return succeed(True)
def describe_addresses(self, *addresses):
return succeed(self.addresses)
def allocate_address(self):
return succeed(self.addresses[0][0])
def release_address(self, address):
return succeed(True)
def associate_address(self, instance_id, address):
return succeed(True)
def disassociate_address(self, address):
return succeed(True)
def describe_availability_zones(self, *names):
return succeed(self.availability_zones)
class FakePageGetter(object):
def __init__(self, status, payload):
self.status = status
self.payload = payload
def get_page(self, url, *args, **kwds):
return succeed(self.payload)
def get_page_with_exception(self, url, *args, **kwds):
try:
raise Error(self.status, "There's been an error", self.payload)
except:
failure = Failure()
return fail(failure)
|