Repository: pathwaycom/pathway
File: integration_tests/webserver/test_rest_connector.py (Python)

Code before:
            1   | 3      | 1
            """,
            output_path,
            usecols=["sum", "diff"],
            index_col=["key"],
        ),
        10,
    )

    t.join()


def test_server_fail_on_duplicate_port(tmp_path: pathlib.Path, port: int) -> None:
    output_path = tmp_path / "output.csv"

    class InputSchema(pw.Schema):
        k: int
        v: int

    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
    queries, response_writer = pw.io.http.rest_connector(
        webserver=webserver, schema=InputSchema, delete_completed_queries=False
    )
    response_writer(queries.select(query_id=queries.id, result=pw.this.v))

    with pytest.raises(ValueError) as exc_info:
        queries_dup, response_writer_dup = pw.io.http.rest_connector(
            webserver=webserver, schema=InputSchema, delete_completed_queries=False
        )
    error_message = str(exc_info.value)
    assert "Added route will never be executed" in error_message
    assert "already registered" in error_message


def _test_server_two_endpoints(
    tmp_path: pathlib.Path, port: int, with_cors: bool
) -> None:
    output_path = tmp_path / "output.csv"

    class InputSchema(pw.Schema):
        query: str
        user: str

    def uppercase_logic(queries: pw.Table) -> pw.Table:
        return queries.select(
            query_id=queries.id, result=pw.apply(lambda x: x.upper(), pw.this.query)
        )

    def duplicate_logic(queries: pw.Table) -> pw.Table:
        return queries.select(
            query_id=queries.id, result=pw.apply(lambda x: x + x, pw.this.query)

Review comment (bug):
The error here has a different type, so the test suite fails. Let's align with an actual error type.

```suggestion
    with pytest.raises(RuntimeError) as exc_info:
```

Code after:
            key | sum    | diff
            1   | 1      | 1
            1   | 1      | -1
            1   | 3      | 1
            """,
            output_path,
            usecols=["sum", "diff"],
            index_col=["key"],
        ),
        10,
    )

    t.join()


def test_server_fail_on_duplicate_port(tmp_path: pathlib.Path, port: int) -> None:
    class InputSchema(pw.Schema):
        k: int
        v: int

    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
    queries, response_writer = pw.io.http.rest_connector(
        webserver=webserver, schema=InputSchema, delete_completed_queries=False
    )
    response_writer(queries.select(query_id=queries.id, result=pw.this.v))

    with pytest.raises(RuntimeError, match="Added route will never be executed"):
        queries_dup, response_writer_dup = pw.io.http.rest_connector(
            webserver=webserver, schema=InputSchema, delete_completed_queries=False
        )


def _test_server_two_endpoints(
    tmp_path: pathlib.Path, port: int, with_cors: bool
) -> None:
    output_path = tmp_path / "output.csv"

    class InputSchema(pw.Schema):
        query: str
        user: str

    def uppercase_logic(queries: pw.Table) -> pw.Table:
        return queries.select(
            query_id=queries.id, result=pw.apply(lambda x: x.upper(), pw.this.query)
        )

    def duplicate_logic(queries: pw.Table) -> pw.Table:
        return queries.select(
            query_id=queries.id, result=pw.apply(lambda x: x + x, pw.this.query)
        )