(stream, raw_data, ctx)
| 109 | |
| 110 | |
| 111 | def test_extract_data(stream, raw_data, ctx): |
| 112 | with Session(ctx.engine) as session: |
| 113 | for each in raw_data: |
| 114 | raw_model = stream.raw_model(session) |
| 115 | raw_model.params = raw_data_params(ctx.connection.id, ctx.scope.id) |
| 116 | session.add(raw_model(data=json.dumps(each))) |
| 117 | session.commit() |
| 118 | |
| 119 | gen = stream.extractor.run(ctx) |
| 120 | list(gen) |
| 121 | |
| 122 | tool_models = session.query(DummyToolModel).all() |
| 123 | alice = tool_models[0] |
| 124 | bob = tool_models[1] |
| 125 | assert alice.name == 'alice' |
| 126 | assert alice.id == 1 |
| 127 | |
| 128 | assert bob.name == 'bob' |
| 129 | assert bob.id == 2 |
| 130 | |
| 131 | |
| 132 | def test_convert_data(stream, raw_data, ctx): |
nothing calls this directly
no test coverage detected