We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
All connections tested even though one's been specified
**Describe the bug**
I ran `astro flow validate --connection=<connection_id>` and all connections were tested even though I passed one conn id specifically. 

**Version**
* Astro Runtime: 7.0.0
* Astro CLI: 1.8.3

**To Reproduce**
Steps to reproduce the behavior:
1. Update file `config/default/configuration.yml` as shown below:
```
connections:
  - conn_id: sqlite_conn
    conn_type: sqlite
    host: /Users/magdagultekin/magda-dev/data/imdb.db
    login: null
    password: null
    schema: null
  - conn_id: sqlite_default
    conn_type: sqlite
    host: /tmp/sqlite.db
    login: null
    password: null
    schema: null
```
4. Run `astro flow validate --connection=sqlite_default`
5. See message:
```
Validating connection(s) for environment 'default'
Validating connection sqlite_conn               PASSED
Validating connection sqlite_default            FAILED
```

**Expected behavior**
Only `sqlite_default` should be tested.

**Screenshots**
![image](https://user-images.githubusercontent.com/97064715/205905971-7c71dc2d-36c8-44ac-9857-9e98180c5ef4.png)



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sql-cli/sql_cli/connections.py`
Content:
```
1 from __future__ import annotations
2 
3 import os
4 from pathlib import Path
5 
6 from airflow.models import Connection
7 
8 from sql_cli.utils.rich import rprint
9 
10 CONNECTION_ID_OUTPUT_STRING_WIDTH = 25
11 
12 
13 def validate_connections(connections: list[Connection], connection_id: str | None = None) -> None:
14     """
15     Validates that the given connections are valid and registers them to Airflow with replace policy for existing
16     connections.
17     """
18     for connection in connections:
19         os.environ[f"AIRFLOW_CONN_{connection.conn_id.upper()}"] = connection.get_uri()
20         status = "[bold green]PASSED[/bold green]" if _is_valid(connection) else "[bold red]FAILED[/bold red]"
21         rprint(f"Validating connection {connection.conn_id:{CONNECTION_ID_OUTPUT_STRING_WIDTH}}", status)
22 
23     if connection_id and not any(connection.conn_id == connection_id for connection in connections):
24         rprint("[bold red]Error: Config file does not contain given connection[/bold red]", connection_id)
25 
26 
27 def _is_valid(connection: Connection) -> bool:
28     # Sqlite automatically creates the file if it does not exist,
29     # but our users might not expect that. They are referencing a database they expect to exist.
30     if connection.conn_type == "sqlite" and not Path(connection.host).is_file():
31         return False
32 
33     success_status, _ = connection.test_connection()
34     return success_status
35
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sql-cli/sql_cli/connections.py b/sql-cli/sql_cli/connections.py
--- a/sql-cli/sql_cli/connections.py
+++ b/sql-cli/sql_cli/connections.py
@@ -15,14 +15,17 @@
     Validates that the given connections are valid and registers them to Airflow with replace policy for existing
     connections.
     """
-    for connection in connections:
-        os.environ[f"AIRFLOW_CONN_{connection.conn_id.upper()}"] = connection.get_uri()
-        status = "[bold green]PASSED[/bold green]" if _is_valid(connection) else "[bold red]FAILED[/bold red]"
-        rprint(f"Validating connection {connection.conn_id:{CONNECTION_ID_OUTPUT_STRING_WIDTH}}", status)
-
     if connection_id and not any(connection.conn_id == connection_id for connection in connections):
         rprint("[bold red]Error: Config file does not contain given connection[/bold red]", connection_id)
 
+    for connection in connections:
+        if not connection_id or connection_id and connection.conn_id == connection_id:
+            os.environ[f"AIRFLOW_CONN_{connection.conn_id.upper()}"] = connection.get_uri()
+            status = (
+                "[bold green]PASSED[/bold green]" if _is_valid(connection) else "[bold red]FAILED[/bold red]"
+            )
+            rprint(f"Validating connection {connection.conn_id:{CONNECTION_ID_OUTPUT_STRING_WIDTH}}", status)
+
 
 def _is_valid(connection: Connection) -> bool:
     # Sqlite automatically creates the file if it does not exist,

{"golden_diff": "diff --git a/sql-cli/sql_cli/connections.py b/sql-cli/sql_cli/connections.py\n--- a/sql-cli/sql_cli/connections.py\n+++ b/sql-cli/sql_cli/connections.py\n@@ -15,14 +15,17 @@\n     Validates that the given connections are valid and registers them to Airflow with replace policy for existing\n     connections.\n     \"\"\"\n-    for connection in connections:\n-        os.environ[f\"AIRFLOW_CONN_{connection.conn_id.upper()}\"] = connection.get_uri()\n-        status = \"[bold green]PASSED[/bold green]\" if _is_valid(connection) else \"[bold red]FAILED[/bold red]\"\n-        rprint(f\"Validating connection {connection.conn_id:{CONNECTION_ID_OUTPUT_STRING_WIDTH}}\", status)\n-\n     if connection_id and not any(connection.conn_id == connection_id for connection in connections):\n         rprint(\"[bold red]Error: Config file does not contain given connection[/bold red]\", connection_id)\n \n+    for connection in connections:\n+        if not connection_id or connection_id and connection.conn_id == connection_id:\n+            os.environ[f\"AIRFLOW_CONN_{connection.conn_id.upper()}\"] = connection.get_uri()\n+            status = (\n+                \"[bold green]PASSED[/bold green]\" if _is_valid(connection) else \"[bold red]FAILED[/bold red]\"\n+            )\n+            rprint(f\"Validating connection {connection.conn_id:{CONNECTION_ID_OUTPUT_STRING_WIDTH}}\", status)\n+\n \n def _is_valid(connection: Connection) -> bool:\n     # Sqlite automatically creates the file if it does not exist,\n", "issue": "All connections tested even though one's been specified\n**Describe the bug**\r\nI ran `astro flow validate --connection=<connection_id>` and all connections were tested even though I passed one conn id specifically. \r\n\r\n**Version**\r\n* Astro Runtime: 7.0.0\r\n* Astro CLI: 1.8.3\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior:\r\n1. Update file `config/default/configuration.yml` as shown below:\r\n```\r\nconnections:\r\n  - conn_id: sqlite_conn\r\n    conn_type: sqlite\r\n    host: /Users/magdagultekin/magda-dev/data/imdb.db\r\n    login: null\r\n    password: null\r\n    schema: null\r\n  - conn_id: sqlite_default\r\n    conn_type: sqlite\r\n    host: /tmp/sqlite.db\r\n    login: null\r\n    password: null\r\n    schema: null\r\n```\r\n4. Run `astro flow validate --connection=sqlite_default`\r\n5. See message:\r\n```\r\nValidating connection(s) for environment 'default'\r\nValidating connection sqlite_conn               PASSED\r\nValidating connection sqlite_default            FAILED\r\n```\r\n\r\n**Expected behavior**\r\nOnly `sqlite_default` should be tested.\r\n\r\n**Screenshots**\r\n![image](https://user-images.githubusercontent.com/97064715/205905971-7c71dc2d-36c8-44ac-9857-9e98180c5ef4.png)\r\n\r\n\n", "before_files": [{"content": "from __future__ import annotations\n\nimport os\nfrom pathlib import Path\n\nfrom airflow.models import Connection\n\nfrom sql_cli.utils.rich import rprint\n\nCONNECTION_ID_OUTPUT_STRING_WIDTH = 25\n\n\ndef validate_connections(connections: list[Connection], connection_id: str | None = None) -> None:\n    \"\"\"\n    Validates that the given connections are valid and registers them to Airflow with replace policy for existing\n    connections.\n    \"\"\"\n    for connection in connections:\n        os.environ[f\"AIRFLOW_CONN_{connection.conn_id.upper()}\"] = connection.get_uri()\n        status = \"[bold green]PASSED[/bold green]\" if _is_valid(connection) else \"[bold red]FAILED[/bold red]\"\n        rprint(f\"Validating connection {connection.conn_id:{CONNECTION_ID_OUTPUT_STRING_WIDTH}}\", status)\n\n    if connection_id and not any(connection.conn_id == connection_id for connection in connections):\n        rprint(\"[bold red]Error: Config file does not contain given connection[/bold red]\", connection_id)\n\n\ndef _is_valid(connection: Connection) -> bool:\n    # Sqlite automatically creates the file if it does not exist,\n    # but our users might not expect that. They are referencing a database they expect to exist.\n    if connection.conn_type == \"sqlite\" and not Path(connection.host).is_file():\n        return False\n\n    success_status, _ = connection.test_connection()\n    return success_status\n", "path": "sql-cli/sql_cli/connections.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport os\nfrom pathlib import Path\n\nfrom airflow.models import Connection\n\nfrom sql_cli.utils.rich import rprint\n\nCONNECTION_ID_OUTPUT_STRING_WIDTH = 25\n\n\ndef validate_connections(connections: list[Connection], connection_id: str | None = None) -> None:\n    \"\"\"\n    Validates that the given connections are valid and registers them to Airflow with replace policy for existing\n    connections.\n    \"\"\"\n    if connection_id and not any(connection.conn_id == connection_id for connection in connections):\n        rprint(\"[bold red]Error: Config file does not contain given connection[/bold red]\", connection_id)\n\n    for connection in connections:\n        if not connection_id or connection_id and connection.conn_id == connection_id:\n            os.environ[f\"AIRFLOW_CONN_{connection.conn_id.upper()}\"] = connection.get_uri()\n            status = (\n                \"[bold green]PASSED[/bold green]\" if _is_valid(connection) else \"[bold red]FAILED[/bold red]\"\n            )\n            rprint(f\"Validating connection {connection.conn_id:{CONNECTION_ID_OUTPUT_STRING_WIDTH}}\", status)\n\n\ndef _is_valid(connection: Connection) -> bool:\n    # Sqlite automatically creates the file if it does not exist,\n    # but our users might not expect that. They are referencing a database they expect to exist.\n    if connection.conn_type == \"sqlite\" and not Path(connection.host).is_file():\n        return False\n\n    success_status, _ = connection.test_connection()\n    return success_status\n", "path": "sql-cli/sql_cli/connections.py"}]}