test_dynamo.py 文件源码

def test_insert_post(self):
        api_res = {'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': 'd92c4314-439a-4bc7-90f5-125a615dfaa2'}}
        db = await self.client.db
        db.put_item = asynctest.CoroutineMock(return_value=api_res)
        date = datetime.utcnow()
        date_str = datetime.strftime(date, self.client.date_fmt)
        kwargs = {"target_id": "target-id",
                  "post_id": "post-id",
                  "source_id": "source-id",
                  "text": "Text",
                  "sticky": True,
                  "created": date,
                  "updated": date,
                  "target_doc": {"foo": "doc"}}
        res = await self.client.insert_post(**kwargs)
        assert res is True
        assert type(res) == bool
        db.put_item.assert_called_once_with(
            Item={
                'source_id': {'S': 'source-id'},
                'text': {'S': 'Text'},
                'target_id': {'S': 'target-id'},
                'post_id': {'S': 'post-id'},
                'created': {'S': date_str},
                'updated': {'S': date_str},
                'target_doc': {'S': '{"foo": "doc"}'},
                'sticky': {'N': '1'}},
            TableName='livebridge_test')

        # insert_post failing(self):
        db.put_item = asynctest.CoroutineMock(side_effect=ParamValidationError(report="Exception raised"))
        res = await self.client.insert_post(**kwargs)
        assert res is False
        assert type(res) == bool