Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: Explicitly add credentials to ThreadedMotoServer tests (#6081)

Browse Source
This commit is contained in:
Bert Blommers 2023-03-17 12:05:05 -01:00 committed by GitHub
parent 0fb00eda5f
commit c4c5cdd3e7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 78 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
tests/test_moto_api/recorder/test_recorder.py
View File

@ -65,14 +65,18 @@ class TestRecorder(TestCase):
self._stop_recording() self._stop_recording()
def test_ec2_instance_creation__recording_off(self): def test_ec2_instance_creation__recording_off(self):
ec2 = boto3.client("ec2", region_name="us-west-1") ec2 = boto3.client(
"ec2", "us-west-1", aws_access_key_id="ak", aws_secret_access_key="sk"
)
ec2.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1) ec2.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
self._download_recording().should.be.empty self._download_recording().should.be.empty
def test_ec2_instance_creation_recording_on(self): def test_ec2_instance_creation_recording_on(self):
self._start_recording() self._start_recording()
ec2 = boto3.client("ec2", region_name="us-west-1") ec2 = boto3.client(
"ec2", "us-west-1", aws_access_key_id="ak", aws_secret_access_key="sk"
)
ec2.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1) ec2.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
content = json.loads(self._download_recording()) content = json.loads(self._download_recording())
@ -82,7 +86,9 @@ class TestRecorder(TestCase):
def test_multiple_services(self): def test_multiple_services(self):
self._start_recording() self._start_recording()
ddb = boto3.client("dynamodb", "eu-west-1") ddb = boto3.client(
"dynamodb", "eu-west-1", aws_access_key_id="ak", aws_secret_access_key="sk"
)
ddb.create_table( ddb.create_table(
TableName="test", TableName="test",
AttributeDefinitions=[{"AttributeName": "client", "AttributeType": "S"}], AttributeDefinitions=[{"AttributeName": "client", "AttributeType": "S"}],
@ -91,10 +97,20 @@ class TestRecorder(TestCase):
) )
ddb.put_item(TableName="test", Item={"client": {"S": "test1"}}) ddb.put_item(TableName="test", Item={"client": {"S": "test1"}})
ts = boto3.client("timestream-write", region_name="us-east-1") ts = boto3.client(
"timestream-write",
"us-east-1",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
ts.create_database(DatabaseName="mydatabase") ts.create_database(DatabaseName="mydatabase")
apigw = boto3.client("apigateway", region_name="us-west-2") apigw = boto3.client(
"apigateway",
"us-west-2",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
apigw.create_rest_api(name="my_api", description="desc") apigw.create_rest_api(name="my_api", description="desc")
content = self._download_recording() content = self._download_recording()
@ -107,10 +123,17 @@ class TestRecorder(TestCase):
def test_replay(self): def test_replay(self):
self._start_recording() self._start_recording()
ddb = boto3.client("dynamodb", "eu-west-1") ddb = boto3.client(
"dynamodb", "eu-west-1", aws_access_key_id="ak", aws_secret_access_key="sk"
)
self._create_ddb_table(ddb, "test") self._create_ddb_table(ddb, "test")
apigw = boto3.client("apigateway", region_name="us-west-2") apigw = boto3.client(
"apigateway",
"us-west-2",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
api_id = apigw.create_rest_api(name="my_api", description="desc")["id"] api_id = apigw.create_rest_api(name="my_api", description="desc")["id"]
self._stop_recording() self._stop_recording()
@ -130,10 +153,17 @@ class TestRecorder(TestCase):
def test_replay__partial_delete(self): def test_replay__partial_delete(self):
self._start_recording() self._start_recording()
ddb = boto3.client("dynamodb", "eu-west-1") ddb = boto3.client(
"dynamodb", "eu-west-1", aws_access_key_id="ak", aws_secret_access_key="sk"
)
self._create_ddb_table(ddb, "test") self._create_ddb_table(ddb, "test")
apigw = boto3.client("apigateway", region_name="us-west-2") apigw = boto3.client(
"apigateway",
"us-west-2",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
api_id = apigw.create_rest_api(name="my_api", description="desc")["id"] api_id = apigw.create_rest_api(name="my_api", description="desc")["id"]
ddb.delete_table(TableName="test") ddb.delete_table(TableName="test")
@ -152,7 +182,9 @@ class TestRecorder(TestCase):
def test_s3_upload_data(self): def test_s3_upload_data(self):
self._start_recording() self._start_recording()
s3 = boto3.client("s3", region_name="us-east-1") s3 = boto3.client(
"s3", "us-east-1", aws_access_key_id="ak", aws_secret_access_key="sk"
)
s3.create_bucket(Bucket="mybucket") s3.create_bucket(Bucket="mybucket")
s3.put_object(Bucket="mybucket", Body=b"ABCD", Key="data") s3.put_object(Bucket="mybucket", Body=b"ABCD", Key="data")
@ -166,7 +198,9 @@ class TestRecorder(TestCase):
resp["Body"].read().should.equal(b"ABCD") resp["Body"].read().should.equal(b"ABCD")
def test_s3_upload_file_using_requests(self): def test_s3_upload_file_using_requests(self):
s3 = boto3.client("s3", region_name="us-east-1") s3 = boto3.client(
"s3", "us-east-1", aws_access_key_id="ak", aws_secret_access_key="sk"
)
s3.create_bucket(Bucket="mybucket") s3.create_bucket(Bucket="mybucket")
params = {"Bucket": "mybucket", "Key": "file_upload"} params = {"Bucket": "mybucket", "Key": "file_upload"}
@ -222,6 +256,8 @@ class TestThreadedMotoServer(TestCase):
"s3", "s3",
region_name="us-east-1", region_name="us-east-1",
endpoint_url=f"http://localhost:{self.port_1}", endpoint_url=f"http://localhost:{self.port_1}",
aws_access_key_id="ak",
aws_secret_access_key="sk",
) )
s3.create_bucket(Bucket="mybucket") s3.create_bucket(Bucket="mybucket")
s3.put_object(Bucket="mybucket", Body=b"ABCD", Key="data") s3.put_object(Bucket="mybucket", Body=b"ABCD", Key="data")
@ -256,6 +292,8 @@ class TestThreadedMotoServer(TestCase):
"s3", "s3",
region_name="us-east-1", region_name="us-east-1",
endpoint_url=f"http://localhost:{self.port_2}", endpoint_url=f"http://localhost:{self.port_2}",
aws_access_key_id="ak",
aws_secret_access_key="sk",
) )
resp = s3.get_object(Bucket="mybucket", Key="data") resp = s3.get_object(Bucket="mybucket", Key="data")
resp["Body"].read().should.equal(b"ABCD") resp["Body"].read().should.equal(b"ABCD")

6
tests/test_s3/test_s3_bucket_policy.py
View File

@ -16,7 +16,11 @@ class TestBucketPolicy:
def setup_method(self) -> None: def setup_method(self) -> None:
self.client = boto3.client( self.client = boto3.client(
"s3", region_name="us-east-1", endpoint_url="http://localhost:6000" "s3",
region_name="us-east-1",
endpoint_url="http://localhost:6000",
aws_access_key_id="ak",
aws_secret_access_key="sk",
) )
self.client.create_bucket(Bucket="mybucket") self.client.create_bucket(Bucket="mybucket")
self.client.put_object(Bucket="mybucket", Key="test_txt", Body=b"mybytes") self.client.put_object(Bucket="mybucket", Key="test_txt", Body=b"mybytes")

28
tests/test_utilities/test_threaded_server.py
View File

@ -19,14 +19,24 @@ class TestThreadedMotoServer(unittest.TestCase):
self.server.stop() self.server.stop()
def test_server_is_reachable(self): def test_server_is_reachable(self):
s3_client = boto3.client("s3", endpoint_url="http://127.0.0.1:5000") s3_client = boto3.client(
"s3",
endpoint_url="http://127.0.0.1:5000",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
s3_client.create_bucket(Bucket="test") s3_client.create_bucket(Bucket="test")
buckets = s3_client.list_buckets()["Buckets"] buckets = s3_client.list_buckets()["Buckets"]
buckets.should.have.length_of(1) buckets.should.have.length_of(1)
[b["Name"] for b in buckets].should.equal(["test"]) [b["Name"] for b in buckets].should.equal(["test"])
def test_server_can_handle_multiple_services(self): def test_server_can_handle_multiple_services(self):
s3_client = boto3.client("s3", endpoint_url="http://127.0.0.1:5000") s3_client = boto3.client(
"s3",
endpoint_url="http://127.0.0.1:5000",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
dynamodb_client = boto3.client( dynamodb_client = boto3.client(
"dynamodb", endpoint_url="http://127.0.0.1:5000", region_name="us-east-1" "dynamodb", endpoint_url="http://127.0.0.1:5000", region_name="us-east-1"
) )
@ -45,7 +55,12 @@ class TestThreadedMotoServer(unittest.TestCase):
@mock_s3 @mock_s3
def test_load_data_from_inmemory_client(self): def test_load_data_from_inmemory_client(self):
server_client = boto3.client("s3", endpoint_url="http://127.0.0.1:5000") server_client = boto3.client(
"s3",
endpoint_url="http://127.0.0.1:5000",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
server_client.create_bucket(Bucket="test") server_client.create_bucket(Bucket="test")
in_mem_client = boto3.client("s3") in_mem_client = boto3.client("s3")
@ -60,7 +75,12 @@ def test_threaded_moto_server__different_port():
server.start() server.start()
requests.post("http://localhost:5001/moto-api/reset") requests.post("http://localhost:5001/moto-api/reset")
try: try:
s3_client = boto3.client("s3", endpoint_url="http://127.0.0.1:5001") s3_client = boto3.client(
"s3",
endpoint_url="http://127.0.0.1:5001",
aws_access_key_id="ak",
aws_secret_access_key="sk",
)
s3_client.create_bucket(Bucket="test") s3_client.create_bucket(Bucket="test")
buckets = s3_client.list_buckets()["Buckets"] buckets = s3_client.list_buckets()["Buckets"]
[b["Name"] for b in buckets].should.equal(["test"]) [b["Name"] for b in buckets].should.equal(["test"])