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

Code before:

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

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

    def target() -> None:
        time.sleep(5)
        r = requests.post(
            f"http://127.0.0.1:{port}",
            json={"query": "one", "user": "sergey"},
        )
        r.raise_for_status()
        assert r.text == '"ONE"', r.text
        r = requests.post(
            f"http://127.0.0.1:{port}",
            json={"query": "two", "user": "sergey"},
        )
        r.raise_for_status()
        assert r.text == '"TWO"', r.text

    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=True
    )
    responses = logic(queries)
    response_writer(responses)
    pw.io.csv.write(queries, output_path)

    t = ExceptionAwareThread(target=target, daemon=True)
    t.start()
    wait_result_with_checker(CsvLinesNumberChecker(output_path, 4), 30)
    t.join()


def test_server(tmp_path: pathlib.Path, port: int) -> None:
    _test_server_basic(tmp_path, port)


def test_server_str_port_compatibility(tmp_path: pathlib.Path, port: int) -> None:
    _test_server_basic(tmp_path, port=str(port))


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

    class InputSchema(pw.Schema):

Review comment (bug):
The `pw.io.http.PathwayWebserver` object only allows an integer as a port. At the same time, `test_server_str_port_compatibility` tests that the strings are also allowed (a legacy of the old code base).

I think we need to restrict ports only to integers, but then the test `test_server_str_port_compatibility` must be removed, otherwise the test suite fails.

Code after:

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

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

    def target() -> None:
        time.sleep(5)
        r = requests.post(
            f"http://127.0.0.1:{port}",
            json={"query": "one", "user": "sergey"},
        )
        r.raise_for_status()
        assert r.text == '"ONE"', r.text
        r = requests.post(
            f"http://127.0.0.1:{port}",
            json={"query": "two", "user": "sergey"},
        )
        r.raise_for_status()
        assert r.text == '"TWO"', r.text

    if isinstance(port, str):
        queries, response_writer = pw.io.http.rest_connector(
            host="127.0.0.1",
            port=port,
            schema=InputSchema,
            delete_completed_queries=True,
        )
    else:
        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=True
        )
    responses = logic(queries)
    response_writer(responses)
    pw.io.csv.write(queries, output_path)

    t = ExceptionAwareThread(target=target, daemon=True)
    t.start()
    wait_result_with_checker(CsvLinesNumberChecker(output_path, 4), 30)
    t.join()


def test_server(tmp_path: pathlib.Path, port: int) -> None:
    _test_server_basic(tmp_path, port)