docs/superpowers/plans/2026-05-06-server-port-constraint.md

@@ -1,147 +0,0 @@
-# Server Port Constraint Implementation Plan
-
-> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
-
-**Goal:** Ensure servers cannot share the same port by enforcing uniqueness at the database level and handling the constraint violation in the web UI.
-
-**Architecture:** We will add a unique constraint to the `Server.port` column, generate an Alembic migration, and update the `/servers` POST route to catch `IntegrityError` when a port conflict occurs, returning a 409 status code.
-
-**Tech Stack:** Python, Flask, SQLAlchemy, Alembic, Pytest
-
----
-
-### Task 1: Add Unique Constraint to Server Port
-
-**Files:**
-- Modify: `l4d2web/models.py`
-- Modify: `l4d2web/routes/server_routes.py`
-- Create: `l4d2web/alembic/versions/XXXX_make_server_port_unique.py` (via alembic)
-
-- [ ] **Step 1: Update the database model**
-
-Update `l4d2web/models.py` to add `unique=True` to the `port` column on the `Server` model.
-
-```python
-class Server(Base):
-    __tablename__ = "servers"
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
-    blueprint_id: Mapped[int] = mapped_column(ForeignKey("blueprints.id"), nullable=False)
-    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
-    port: Mapped[int] = mapped_column(Integer, unique=True, nullable=False)
-    # ... rest of the columns
-```
-
-- [ ] **Step 2: Generate the Alembic migration**
-
-Run: `PYTHONPATH=. alembic -c l4d2web/alembic.ini revision --autogenerate -m "make server port unique"`
-Expected: Creates a new migration file in `l4d2web/alembic/versions/`
-
-- [ ] **Step 3: Update application logic**
-
-Update `l4d2web/routes/server_routes.py` to catch the `IntegrityError` when creating a server.
-
-```python
-from sqlalchemy.exc import IntegrityError
-# ... other imports
-
-@bp.post("/servers")
-@require_login
-def create_server() -> Response:
-    # ... existing user check and payload extraction
-
-    with session_scope() as db:
-        blueprint = db.scalar(
-            select(BlueprintModel).where(
-                BlueprintModel.id == int(payload["blueprint_id"]),
-                BlueprintModel.user_id == user.id,
-            )
-        )
-        if blueprint is None:
-            return Response("blueprint not found", status=404)
-
-        server = Server(
-            user_id=user.id,
-            blueprint_id=blueprint.id,
-            name=str(payload["name"]),
-            port=int(payload["port"]),
-            desired_state="stopped",
-            actual_state="unknown",
-            last_error="",
-        )
-        db.add(server)
-        
-        try:
-            db.flush()
-        except IntegrityError:
-            db.rollback()
-            return Response("port already in use", status=409)
-            
-        server_id = server.id
-
-    if json_response:
-        return jsonify({"id": server_id}), 201
-    return redirect(f"/servers/{server_id}")
-```
-
-- [ ] **Step 4: Commit**
-
-```bash
-git add l4d2web/models.py l4d2web/routes/server_routes.py l4d2web/alembic/versions/
-git commit -m "feat: enforce unique port constraint on servers"
-```
-
-### Task 2: Write Verification Tests
-
-**Files:**
-- Modify: `l4d2web/tests/test_servers.py`
-
-- [ ] **Step 1: Write the failing test**
-
-Add a test case to `l4d2web/tests/test_servers.py` that verifies the unique port constraint.
-
-```python
-def test_create_server_duplicate_port(client, auth, db_session):
-    auth.login()
-    
-    # First, create a blueprint
-    response = client.post(
-        "/blueprints",
-        data={"name": "my-blueprint", "arguments": "[]", "config": "[]"},
-    )
-    assert response.status_code == 302
-    
-    # Then create the first server
-    response = client.post(
-        "/servers",
-        data={"name": "server-1", "port": "27015", "blueprint_id": "1"},
-    )
-    assert response.status_code == 302
-    
-    # Try to create a second server with the same port
-    response = client.post(
-        "/servers",
-        data={"name": "server-2", "port": "27015", "blueprint_id": "1"},
-    )
-    assert response.status_code == 409
-    assert b"port already in use" in response.data
-    
-    # Verify the second server was not created
-    from l4d2web.models import Server
-    servers = db_session.query(Server).all()
-    assert len(servers) == 1
-    assert servers[0].name == "server-1"
-```
-
-- [ ] **Step 2: Run test to verify it passes**
-
-Run: `pytest l4d2web/tests/test_servers.py -v`
-Expected: PASS (It passes because we implemented the code in Task 1. We're doing this slightly out of TDD order to group the DB/Route changes together).
-
-- [ ] **Step 3: Commit**
-
-```bash
-git add l4d2web/tests/test_servers.py
-git commit -m "test: add test for duplicate port constraint"
-```

l4d2web/alembic/env.py

@@ -25,7 +25,6 @@ def run_migrations_offline() -> None:
         target_metadata=target_metadata,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
-        render_as_batch=True
     )
 
     with context.begin_transaction():
@@ -42,11 +41,7 @@ def run_migrations_online() -> None:
     )
 
     with connectable.connect() as connection:
-        context.configure(
+        context.configure(connection=connection, target_metadata=target_metadata)
-            connection=connection, 
-            target_metadata=target_metadata,
-            render_as_batch=True
-        )
 
         with context.begin_transaction():
             context.run_migrations()

l4d2web/alembic/script.py.mako

@@ -1,28 +0,0 @@
-"""${message}
-
-Revision ID: ${up_revision}
-Revises: ${down_revision | comma,n}
-Create Date: ${create_date}
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-${imports if imports else ""}
-
-# revision identifiers, used by Alembic.
-revision: str = ${repr(up_revision)}
-down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
-branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
-depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
-
-
-def upgrade() -> None:
-    """Upgrade schema."""
-    ${upgrades if upgrades else "pass"}
-
-
-def downgrade() -> None:
-    """Downgrade schema."""
-    ${downgrades if downgrades else "pass"}

l4d2web/alembic/versions/b2c684fddbd3_make_server_port_unique.py

@@ -1,34 +0,0 @@
-"""make server port unique
-
-Revision ID: b2c684fddbd3
-Revises: 0001_initial
-Create Date: 2026-05-06 20:52:35.109176
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = 'b2c684fddbd3'
-down_revision: Union[str, Sequence[str], None] = '0001_initial'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    """Upgrade schema."""
-    # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('servers', schema=None) as batch_op:
-        batch_op.create_unique_constraint(batch_op.f('uq_servers_port'), ['port'])
-    # ### end Alembic commands ###
-
-
-def downgrade() -> None:
-    """Downgrade schema."""
-    # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('servers', schema=None) as batch_op:
-        batch_op.drop_constraint(batch_op.f('uq_servers_port'), type_='unique')
-    # ### end Alembic commands ###

l4d2web/models.py

@@ -63,7 +63,7 @@ class Server(Base):
     user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
     blueprint_id: Mapped[int] = mapped_column(ForeignKey("blueprints.id"), nullable=False)
     name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
-    port: Mapped[int] = mapped_column(Integer, unique=True, nullable=False)
+    port: Mapped[int] = mapped_column(Integer, nullable=False)
     desired_state: Mapped[str] = mapped_column(String(16), default="stopped", nullable=False)
     actual_state: Mapped[str] = mapped_column(String(16), default="unknown", nullable=False)
     actual_state_updated_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

l4d2web/routes/server_routes.py

@@ -1,6 +1,5 @@
 from flask import Blueprint, Response, jsonify, redirect, request
 from sqlalchemy import select
-from sqlalchemy.exc import IntegrityError
 
 from l4d2web.auth import current_user, require_login
 from l4d2web.db import session_scope
@@ -39,13 +38,7 @@ def create_server() -> Response:
             last_error="",
         )
         db.add(server)
-        
+        db.flush()
-        try:
-            db.flush()
-        except IntegrityError:
-            db.rollback()
-            return Response("port already in use", status=409)
-            
         server_id = server.id
 
     if json_response:

l4d2web/tests/test_servers.py

@@ -83,37 +83,6 @@ def test_reassign_blueprint_anytime(user_client_with_blueprints) -> None:
     assert response.status_code == 200
 
 
-def test_create_server_duplicate_port(user_client_with_blueprints) -> None:
-    client, data = user_client_with_blueprints
-    
-    # Create the first server
-    response = client.post(
-        "/servers",
-        data={"name": "server-1", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
-        headers={"X-CSRF-Token": "test-token"},
-    )
-    assert response.status_code == 302
-    
-    # Try to create a second server with the same port
-    response = client.post(
-        "/servers",
-        data={"name": "server-2", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
-        headers={"X-CSRF-Token": "test-token"},
-    )
-    assert response.status_code == 409
-    assert b"port already in use" in response.data
-    
-    # Verify the second server was not created by checking how many servers exist
-    from sqlalchemy import select
-    from l4d2web.db import session_scope
-    from l4d2web.models import Server
-    
-    with session_scope() as session:
-        servers = session.scalars(select(Server)).all()
-        assert len(servers) == 1
-        assert servers[0].name == "server-1"
-
-
 def test_lifecycle_form_creates_queued_job(user_client_with_blueprints) -> None:
     client, data = user_client_with_blueprints
     create_response = client.post(