(stream, raw_data, ctx)
| 130 | |
| 131 | |
| 132 | def test_convert_data(stream, raw_data, ctx): |
| 133 | with Session(ctx.engine) as session: |
| 134 | for each in raw_data: |
| 135 | session.add( |
| 136 | DummyToolModel( |
| 137 | id=each["i"], |
| 138 | connection_id=ctx.connection.id, |
| 139 | name=each["n"], |
| 140 | raw_data_table="_raw_dummy_model", |
| 141 | raw_data_params=raw_data_params(ctx.connection.id, ctx.scope.id) |
| 142 | ) |
| 143 | ) |
| 144 | session.commit() |
| 145 | |
| 146 | gen = stream.convertor.run(ctx) |
| 147 | list(gen) |
| 148 | |
| 149 | tool_models = session.query(DummyDomainModel).all() |
| 150 | alice = tool_models[0] |
| 151 | bob = tool_models[1] |
| 152 | assert alice.Name == 'alice' |
| 153 | assert alice.id == 'tests:DummyToolModel:11:1' |
| 154 | |
| 155 | assert bob.Name == 'bob' |
| 156 | assert bob.id == 'tests:DummyToolModel:11:2' |
nothing calls this directly
no test coverage detected