Validate connection port is a valid network port number

[dj-wise-ronin]

Description

Connection models in Apache Airflow accept integer values for the port field but do not enforce that the value is a valid network port in the range [1, 65535].
This PR adds validation logic to connection ports at three layers:

1. SQLAlchemy Connection model validation via @validates("port") to ensure port numbers assigned are integers between 1 and 65535.
2. Task SDK Connection model validation in __attrs_post_init__ to enforce range limits and format validation during standard instantiation and URI parsing.
3. Pydantic API datamodel validation using Field(ge=1, le=65535) on the REST API gateway connections schema.

Additionally, this PR includes comprehensive unit tests verifying the validation rules for both the SQLAlchemy and Task SDK model implementations.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[SameerMesiah97]

I have left comments on the test.

Also, I noticed the Task SDK and SQLAlchemy models now contain identical port parsing/validation logic. Is there an opportunity to share the implementation?

[SameerMesiah97]

The tests could be better. Boundary and whitespace coverage is missing amongst other things. I would suggest the below:

class TestConnectionPortValidation:
    """Test connection port validation in the task-sdk."""

    @pytest.mark.parametrize(
        ("port", "expected"),
        [
            pytest.param(1, 1, id="min_port"),
            pytest.param(80, 80, id="valid_port"),
            pytest.param("8080", 8080, id="string_port"),
            pytest.param(65535, 65535, id="max_port"),
            pytest.param(None, None, id="none"),
            pytest.param("", None, id="empty_string"),
            pytest.param("   ", None, id="whitespace"),
        ],
    )
    def test_valid_port(self, port, expected):
        """Test valid port values."""
        assert Connection(conn_id="test", port=port).port == expected

    @pytest.mark.parametrize(
        "port",
        [
            pytest.param(0, id="zero"),
            pytest.param(-80, id="negative"),
            pytest.param(70000, id="too_large"),
        ],
    )
    def test_invalid_port_range(self, port):
        """Test ports outside the valid network port range."""
        with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
            Connection(conn_id="test", port=port)

    @pytest.mark.parametrize(
        "port",
        [
            pytest.param("invalid_port", id="invalid_string"),
        ],
    )
    def test_invalid_port_type(self, port):
        """Test non-integer port values."""
        with pytest.raises(ValueError, match="Expected integer value for `port`"):
            Connection(conn_id="test", port=port)

    def test_from_uri_port_validation(self):
        """Test that Connection.from_uri validates the port field correctly."""
        assert Connection.from_uri("postgres://host:5432/db", conn_id="test").port == 5432

        with pytest.raises(ValueError, match="Port out of range 0-65535"):
            Connection.from_uri("postgres://host:70000/db", conn_id="test")

        with pytest.raises(ValueError, match="Port could not be cast to integer value"):
            Connection.from_uri("postgres://host:abc/db", conn_id="test")

[SameerMesiah97]

This test could be better as well. Please see the below:

@pytest.mark.parametrize(
    ("port", "expected"),
    [
        pytest.param(1, 1, id="min_port"),
        pytest.param(80, 80, id="valid_port"),
        pytest.param(65535, 65535, id="max_port"),
        pytest.param("8080", 8080, id="string_port"),
        pytest.param(None, None, id="none"),
        pytest.param("", None, id="empty_string"),
        pytest.param("  ", None, id="whitespace"),
    ],
)
def test_port_validation(self, port, expected):
    """Test that Connection model validates the port field correctly."""
    assert Connection(conn_id="test", port=port).port == expected


@pytest.mark.parametrize(
    "port",
    [
        pytest.param(70000, id="too_large"),
        pytest.param(0, id="zero"),
        pytest.param(-80, id="negative"),
    ],
)
def test_port_validation_out_of_range(self, port):
    """Test that out-of-range ports are rejected."""
    with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
        Connection(conn_id="test", port=port)


@pytest.mark.parametrize(
    "port",
    [
        pytest.param("invalid_port", id="invalid_string"),
    ],
)
def test_port_validation_invalid_type(self, port):
    """Test that invalid port types are rejected."""
    with pytest.raises(ValueError, match="Expected integer value for `port`"):
        Connection(conn_id="test", port=port)

[dj-wise-ronin]

I have refactored the codebase to share the connection port range validation logic between airflow-core (Connection SQLAlchemy model) and task-sdk (Connection class):

• Shared Port Validator: Created a central parse_and_validate_port validator helper within the shared configurations package (airflow_shared.configuration.connection).
• Refactored Models:
  • Core Connection model's @validates("port") method now delegates directly to the shared validator.
  • Task SDK Connection model's __attrs_post_init__ method delegates to the shared validator.
• Tests: Updated and verified that all existing tests in both airflow-core and task-sdk pass successfully.

[github-actions]

uv.lock on main just moved via #70086 ("Fix stale apache-airflow-providers-snowflake provider in generator constraints with cap to cloudpickle"), commit a7a9fed and this PR currently conflicts.

Quickest fix:

git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease

Automated nudge — ignore if you're not ready to rebase. This comment is updated in place on future uv.lock bumps.