ferruzzi commented on PR #28338:
URL: https://github.com/apache/airflow/pull/28338#issuecomment-1496588815
I am neither an expert in mocking nor in Dynamo, but I leaned heavily on the
existing dynamohook tests to write this and it does pass with your Sensor.
Give it a try, use it as a base test, and modify it as needed. :+1:
In a nutshell, instead of mocking the connection and return values, I let
moto handle everything. That's what it is expected to do (normally....) it's
not perfect, but it _is_ OSS and accepts contributions if it's missing an API
call :stuck_out_tongue:
```python
class TestDynamoDBValueSensor:
def setup_method(self):
self.table_name = "test_airflow"
self.key_name = "PK"
self.key_value = "Test"
self.attribute_name = "Foo"
self.attribute_value = "Bar"
self.sensor = DynamoDBValueSensor(
task_id="dynamodb_value_sensor",
table_name=self.table_name,
partition_key_name=self.key_name,
partition_key_value=self.key_value,
attribute_name=self.attribute_name,
attribute_value=self.attribute_value,
)
@mock_dynamodb
def test_sensor_with_pk(self):
hook = DynamoDBHook(table_name=self.table_name,
table_keys=[self.key_name])
hook.conn.create_table(
TableName=self.table_name,
KeySchema=[{"AttributeName": self.key_name, "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": self.key_name,
"AttributeType": "S"}],
ProvisionedThroughput={"ReadCapacityUnits": 10,
"WriteCapacityUnits": 10},
)
items = [{self.key_name: self.key_value, self.attribute_name:
self.attribute_value}]
hook.write_batch_data(items)
assert self.sensor.poke({})
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
