devplacepy/tests/unit/services/containers/api.py

# retoor <retoor@molodetz.nl>

from datetime import datetime, timedelta, timezone
import os
import pytest
import requests
from tests.conftest import BASE_URL, run_async
from devplacepy.database import db, get_table, init_db, refresh_snapshot
from devplacepy.utils import generate_uid
from devplacepy import config, project_files
from starlette.requests import Request
from starlette.websockets import WebSocket
from devplacepy.services.containers import api, forward, store, runtime
from devplacepy.services.containers.backend.base import Mount, PortMapping, RunSpec
from devplacepy.services.containers.backend.docker_cli import build_run_argv, parse_size
from devplacepy.services.containers.backend.fake import FakeBackend
from devplacepy.services.containers.service import ContainerService
from devplacepy.services.containers.api import INSTANCE_LABEL
from devplacepy.services.devii.tasks.schedule import Schedule, now_utc
_CONTAINER_TABLES = (
    "instances",
    "instance_events",
    "instance_metrics",
    "instance_schedules",
)
@pytest.fixture(autouse=True)
def _init_db_containers():
    init_db()
    yield
@pytest.fixture
def env(tmp_path, monkeypatch):
    fake = FakeBackend()
    runtime.set_backend(fake)
    monkeypatch.setattr("devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path / "ws")
    pid = "ctest-p1"
    project = {"uid": pid, "slug": "ctest", "title": "C", "user_uid": "ctest-u1"}
    user = {"uid": "ctest-u1", "username": "ctestadmin"}
    get_table("users").insert(
        {
            "uid": user["uid"],
            "username": user["username"],
            "terms_version": "1",
            "email": "ctestadmin@example.com",
            "api_key": generate_uid(),
            "password_hash": "",
            "role": "Member",
            "is_active": True,
            "level": 1,
            "xp": 0,
            "stars": 0,
            "deleted_at": None,
            "deleted_by": None,
        }
    )
    refresh_snapshot()
    project_files.write_text_file(pid, user, "app.py", "print(1)\n")
    yield {"fake": fake, "project": project, "user": user}
    runtime.set_backend(None)
    get_table("users").delete(uid=user["uid"])
    for table in _CONTAINER_TABLES:
        if table in db.tables:
            for row in [r for r in get_table(table).find()]:
                if str(row.get("project_uid", "")).startswith("ctest"):
                    get_table(table).delete(uid=row["uid"])
    for row in list(get_table("project_files").find()):
        if str(row.get("project_uid", "")).startswith("ctest"):
            get_table("project_files").delete(uid=row["uid"])
    if "project_file_sync_state" in db.tables:
        for row in list(get_table("project_file_sync_state").find()):
            if str(row.get("project_uid", "")).startswith("ctest"):
                get_table("project_file_sync_state").delete(
                    project_uid=row["project_uid"], path=row["path"]
                )
    if "projects" in db.tables:
        get_table("projects").delete(uid=pid)
def _ready_instance(env, **kwargs):
    return run_async(
        api.create_instance(env["project"], name=kwargs.pop("name", "inst"), **kwargs)
    )
def _promote_admin(username: str) -> None:
    users = get_table("users")
    user = users.find_one(username=username)
    if user:
        users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
def _api_key(username: str) -> str:
    refresh_snapshot()
    return get_table("users").find_one(username=username)["api_key"]


def test_build_run_argv_exact():
    spec = RunSpec(
        image="ppy:latest",
        name="inst",
        labels={INSTANCE_LABEL: "u1"},
        env={"A": "B"},
        cpu_limit="1.5",
        mem_limit="512m",
        ports=[PortMapping(8080, 80)],
        mounts=[Mount("/ws", "/app")],
        restart_policy="on-failure",
        command=["python", "app.py"],
    )
    argv = build_run_argv(spec)
    assert argv[:5] == ["docker", "run", "-d", "--name", "inst"]
    assert "--label" in argv and f"{INSTANCE_LABEL}=u1" in argv
    assert "--cpus" in argv and "1.5" in argv
    assert "-p" in argv and "8080:80/tcp" in argv
    assert "-v" in argv and "/ws:/app:rw" in argv
    assert "--restart" in argv and "on-failure" in argv
    assert argv[-3:] == ["ppy:latest", "python", "app.py"]


def test_never_policy_not_passed_to_docker():
    spec = RunSpec(image="ppy:latest", name="n", restart_policy="never")
    assert "--restart" not in build_run_argv(spec)


def test_parse_size():
    assert parse_size("1.0GiB") == 1024**3
    assert parse_size("512MB") == 512 * 1024**2


def test_create_instance_uses_shared_image(env):
    inst = run_async(
        api.create_instance(
            env["project"], name="inst", actor=("user", env["user"]["uid"])
        )
    )
    assert inst["name"] == "inst"
    assert inst["owner_uid"] == "ctest-u1"
    spec = api.run_spec_for(inst, config.CONTAINER_IMAGE)
    assert spec.image == config.CONTAINER_IMAGE
    assert any(m.container == "/app" for m in spec.mounts)


def test_create_instance_requires_built_image(env):
    async def no_image(ref):
        return False

    env["fake"].image_exists = no_image
    with pytest.raises(api.ContainerError):
        run_async(api.create_instance(env["project"], name="inst"))


def test_reconcile_launches_and_stops(env):
    inst = _ready_instance(env, restart_policy="never", autostart=True)
    assert inst["desired_state"] == store.DESIRED_RUNNING
    service = ContainerService()
    run_async(service.run_once())
    refresh_snapshot()
    inst = store.get_instance(inst["uid"])
    assert inst["status"] == store.ST_RUNNING and inst["container_id"]
    assert [r.name for r in run_async(env["fake"].ps())] == [inst["slug"]]
    api.set_desired_state(inst, store.DESIRED_STOPPED)
    run_async(service.run_once())
    refresh_snapshot()
    assert store.get_instance(inst["uid"])["status"] == store.ST_STOPPED


def test_manual_start_relaunches_after_never_policy_exit(env):
    fake = env["fake"]
    inst = _ready_instance(env, restart_policy="never", autostart=True)
    service = ContainerService()
    run_async(service.run_once())
    refresh_snapshot()
    inst = store.get_instance(inst["uid"])
    cid = inst["container_id"]
    assert inst["status"] == store.ST_RUNNING and cid

    fake._set_state(cid, "exited", 0)
    run_async(service.run_once())
    refresh_snapshot()
    inst = store.get_instance(inst["uid"])
    assert inst["status"] == store.ST_STOPPED
    assert inst["desired_state"] == store.DESIRED_STOPPED

    api.set_desired_state(inst, store.DESIRED_RUNNING)
    refresh_snapshot()
    run_async(service.run_once())
    refresh_snapshot()
    inst = store.get_instance(inst["uid"])
    assert inst["status"] == store.ST_RUNNING
    assert inst["desired_state"] == store.DESIRED_RUNNING
    assert cid in fake.removed
    assert inst["container_id"] and inst["container_id"] != cid


def test_reconcile_reaps_orphan(env):
    fake = env["fake"]
    run_async(
        fake.run(
            RunSpec(
                image="ppy:latest", name="ghost", labels={INSTANCE_LABEL: "missing-uid"}
            )
        )
    )
    service = ContainerService()
    run_async(service.run_once())
    assert not run_async(fake.ps())
    assert fake.removed


def test_reconcile_removes_marked_instance(env):
    inst = _ready_instance(env, autostart=True)
    service = ContainerService()
    run_async(service.run_once())
    refresh_snapshot()
    inst = store.get_instance(inst["uid"])
    api.mark_for_removal(inst)
    run_async(service.run_once())
    refresh_snapshot()
    assert store.get_instance(inst["uid"]) is None
    assert not run_async(env["fake"].ps())


def test_schedule_fires(env):
    inst = _ready_instance(env, autostart=False)
    assert inst["desired_state"] == store.DESIRED_STOPPED
    past = Schedule(kind="once", run_at=now_utc() - timedelta(hours=1))
    api.add_schedule(inst, "start", past)
    service = ContainerService()
    run_async(service._fire_schedules())
    refresh_snapshot()
    assert store.get_instance(inst["uid"])["desired_state"] == store.DESIRED_RUNNING


def test_ingress_validation(env):
    from devplacepy.services.containers.backend.base import PortMapping

    assert api.validate_ingress("zwoeks", 8899, [PortMapping(8899, 8899)]) == (
        "zwoeks",
        8899,
    )
    assert api.validate_ingress("", None, []) == ("", 0)
    with pytest.raises(api.ContainerError):
        api.validate_ingress("BAD SLUG", None, [PortMapping(80, 80)])
    with pytest.raises(api.ContainerError):
        api.validate_ingress("x", 9999, [PortMapping(80, 80)])
    store.create_instance(
        {
            "uid": "z",
            "project_uid": "ctest-p1",
            "name": "z",
            "ports_json": "[]",
            "ingress_slug": "taken",
        }
    )
    with pytest.raises(api.ContainerError):
        api.validate_ingress("taken", None, [PortMapping(80, 80)])


def test_validate_boot_languages():
    assert api.validate_boot("none", "ignored") == ("none", "")
    assert api.validate_boot("python", "print(1)") == ("python", "print(1)")
    with pytest.raises(api.ContainerError):
        api.validate_boot("ruby", "puts 1")
    with pytest.raises(api.ContainerError):
        api.validate_boot("bash", "   ")


def test_boot_script_precedence_in_run_spec(env):
    inst = _ready_instance(
        env,
        boot_language="python",
        boot_script="print('boot')\n",
        boot_command="python other.py",
        autostart=False,
    )
    spec = api.run_spec_for(inst, config.CONTAINER_IMAGE)
    assert spec.command == ["python", "/app/.devplace_boot.py"]
    inst2 = _ready_instance(
        env, name="i2", boot_command="python serve.py", autostart=False
    )
    spec2 = api.run_spec_for(inst2, config.CONTAINER_IMAGE)
    assert spec2.command == ["/bin/sh", "-c", "python serve.py"]


def test_materialize_boot_script_writes_file(env, tmp_path):
    workspace = tmp_path / "boot-ws"
    inst = _ready_instance(
        env, name="boot", boot_language="bash", boot_script="echo hi\n", autostart=False
    )
    store.update_instance(inst["uid"], {"workspace_dir": str(workspace)})
    inst = store.get_instance(inst["uid"])
    api.materialize_boot_script(inst)
    written = workspace / ".devplace_boot.sh"
    assert written.is_file()
    assert written.read_text() == "echo hi\n"


def test_run_as_uid_drives_pravda_env(env):
    _promote_admin(env["user"]["username"])
    key = _api_key(env["user"]["username"])
    inst = _ready_instance(
        env, name="runas", run_as_uid=env["user"]["uid"], autostart=False
    )
    pravda = api.pravda_env(inst)
    assert pravda["DEVPLACE_API_KEY"] == key
    assert pravda["DEVPLACE_USER_UID"] == env["user"]["uid"]


def test_run_as_uid_rejects_unknown_user(env):
    with pytest.raises(api.ContainerError):
        run_async(
            api.create_instance(
                env["project"], name="bad", run_as_uid="nope-uid", autostart=False
            )
        )


def _other_member(uid="ctest-u2", username="ctestmember"):
    users = get_table("users")
    if not users.find_one(uid=uid):
        users.insert(
            {
                "uid": uid,
                "username": username,
                "terms_version": "1",
                "email": f"{username}@example.com",
                "api_key": generate_uid(),
                "password_hash": "",
                "role": "Member",
                "is_active": True,
                "created_at": "2020-01-01T00:00:00+00:00",
            }
        )
    refresh_snapshot()
    return uid


def test_run_as_another_member_needs_their_credential_consent(env):
    from devplacepy.database import set_consent

    other = _other_member()
    set_consent("user", other, "container_credentials", False)
    with pytest.raises(api.ContainerError) as refused:
        run_async(
            api.create_instance(
                env["project"],
                name="noconsent",
                run_as_uid=other,
                autostart=False,
                actor=("user", env["user"]["uid"]),
            )
        )
    assert "consent" in str(refused.value).lower()


def test_run_as_another_member_is_allowed_once_they_consent(env):
    from devplacepy.database import set_consent

    other = _other_member()
    set_consent("user", other, "container_credentials", True)
    instance = run_async(
        api.create_instance(
            env["project"],
            name="withconsent",
            run_as_uid=other,
            autostart=False,
            actor=("user", env["user"]["uid"]),
        )
    )
    assert instance["run_as_uid"] == other


def test_run_as_yourself_never_needs_a_consent(env):
    instance = run_async(
        api.create_instance(
            env["project"],
            name="selfrunas",
            run_as_uid=env["user"]["uid"],
            autostart=False,
            actor=("user", env["user"]["uid"]),
        )
    )
    assert instance["run_as_uid"] == env["user"]["uid"]


def test_start_on_boot_pass_forces_running(env):
    inst = _ready_instance(env, name="boot", start_on_boot=True, autostart=False)
    assert inst["desired_state"] == store.DESIRED_STOPPED
    service = ContainerService()
    service._boot_pass(store.all_instances())
    refresh_snapshot()
    assert store.get_instance(inst["uid"])["desired_state"] == store.DESIRED_RUNNING


def test_status_change_records_event(env):
    inst = _ready_instance(env, name="hist", restart_policy="never", autostart=True)
    service = ContainerService()
    run_async(service.run_once())
    refresh_snapshot()
    inst = store.get_instance(inst["uid"])
    api.set_desired_state(inst, store.DESIRED_STOPPED)
    run_async(service.run_once())
    refresh_snapshot()
    events = [e["event"] for e in store.list_events(inst["uid"])]
    assert "status_change" in events


def test_bidirectional_sync_newer_wins(env, tmp_path):
    workspace = tmp_path / "sync-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "shared.txt", "from project\n")
    fs_only = workspace / "fromfs.txt"
    fs_only.write_text("from fs\n")
    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)
    assert counts["exported"] >= 1
    assert counts["imported"] >= 1
    assert (workspace / "shared.txt").read_text() == "from project\n"
    imported = project_files.read_file(pid, "fromfs.txt")
    assert imported["content"] == "from fs\n"


def test_bidirectional_sync_propagates_a_workspace_deletion(env, tmp_path):
    workspace = tmp_path / "sync-del-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "gone.txt", "bye\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)
    assert (workspace / "gone.txt").exists()

    (workspace / "gone.txt").unlink()
    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)

    assert counts["deleted_in_project"] == 1
    assert counts["exported"] == 0
    assert project_files.get_node(pid, "gone.txt") is None
    assert not (workspace / "gone.txt").exists()


def test_bidirectional_sync_propagates_a_project_deletion(env, tmp_path):
    workspace = tmp_path / "sync-pdel-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "removeme.txt", "bye\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)
    assert (workspace / "removeme.txt").exists()

    project_files.delete_node(pid, "removeme.txt", deleted_by=user["uid"])
    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)

    assert counts["deleted_in_workspace"] == 1
    assert counts["imported"] == 0
    assert not (workspace / "removeme.txt").exists()


def test_bidirectional_sync_a_later_edit_restores_a_workspace_deletion(env, tmp_path):
    workspace = tmp_path / "sync-edit-restore-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    node = project_files.write_text_file(pid, user, "edited.txt", "v1\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)

    (workspace / "edited.txt").unlink()
    future = datetime.now(timezone.utc) + timedelta(seconds=10)
    get_table("project_files").update(
        {"uid": node["uid"], "content": "v2\n", "updated_at": future.isoformat()},
        ["uid"],
    )
    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)

    assert counts["deleted_in_project"] == 0
    assert counts["exported"] == 1
    assert (workspace / "edited.txt").read_text() == "v2\n"


def test_bidirectional_sync_a_later_local_edit_reimports_over_a_project_deletion(
    env, tmp_path
):
    workspace = tmp_path / "sync-edit-reimport-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "revived.txt", "v1\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)

    project_files.delete_node(pid, "revived.txt", deleted_by=user["uid"])
    target = workspace / "revived.txt"
    target.write_text("v2\n")
    future = (datetime.now(timezone.utc) + timedelta(seconds=10)).timestamp()
    os.utime(target, (future, future))
    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)

    assert counts["deleted_in_workspace"] == 0
    assert counts["imported"] == 1
    revived = project_files.read_file(pid, "revived.txt")
    assert revived["content"] == "v2\n"


def test_bidirectional_sync_readonly_always_restores_a_workspace_deletion(
    env, tmp_path
):
    workspace = tmp_path / "sync-ro-restore-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "frozen.txt", "kept\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)
    get_table("projects").upsert(
        {"uid": pid, "slug": "ctest", "read_only": 1}, ["uid"]
    )

    (workspace / "frozen.txt").unlink()
    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)

    assert counts["deleted_in_project"] == 0
    assert counts["exported"] == 1
    assert (workspace / "frozen.txt").read_text() == "kept\n"
    assert project_files.get_node(pid, "frozen.txt") is not None


def test_bidirectional_sync_readonly_removes_a_stale_local_copy(env, tmp_path):
    workspace = tmp_path / "sync-ro-remove-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "stale.txt", "stale\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)
    project_files.delete_node(pid, "stale.txt", deleted_by=user["uid"])
    get_table("projects").upsert(
        {"uid": pid, "slug": "ctest", "read_only": 1}, ["uid"]
    )

    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)

    assert counts["deleted_in_workspace"] == 1
    assert counts["imported"] == 0
    assert not (workspace / "stale.txt").exists()


def test_bidirectional_sync_is_a_noop_once_both_sides_settle(env, tmp_path):
    workspace = tmp_path / "sync-noop-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "settled.txt", "steady\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)
    before = (workspace / "settled.txt").stat().st_mtime

    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)

    assert counts == {
        "exported": 0,
        "imported": 0,
        "deleted_in_project": 0,
        "deleted_in_workspace": 0,
    }
    assert (workspace / "settled.txt").stat().st_mtime == before


def test_bidirectional_sync_manifest_is_pruned_after_both_sides_agree(env, tmp_path):
    workspace = tmp_path / "sync-prune-ws"
    workspace.mkdir()
    pid = env["project"]["uid"]
    user = env["user"]
    project_files.write_text_file(pid, user, "prune.txt", "x\n")
    project_files.sync_dir_bidirectional(pid, str(workspace), user)
    (workspace / "prune.txt").unlink()
    project_files.sync_dir_bidirectional(pid, str(workspace), user)

    remaining = list(
        get_table("project_file_sync_state").find(project_uid=pid, path="prune.txt")
    )
    assert remaining == []



def _proxy_scope(kind: str, headers: dict, scheme: str, query: str = "") -> dict:
    return {
        "type": kind,
        "scheme": scheme,
        "server": ("devplace.net", 443),
        "path": "/projects/demo/containers/instances/abc/code/stable-1",
        "query_string": query.encode(),
        "headers": [
            (name.encode(), value.encode()) for name, value in headers.items()
        ],
    }


def _proxy_websocket(headers: dict, scheme: str = "wss", query: str = "") -> WebSocket:
    return WebSocket(_proxy_scope("websocket", headers, scheme, query), None, None)


def _proxy_request(headers: dict, scheme: str = "https", query: str = "") -> Request:
    return Request(_proxy_scope("http", headers, scheme, query))


def test_ws_headers_send_the_public_host_only_as_x_forwarded_host():
    headers = forward.ws_headers(
        _proxy_websocket(
            {
                "host": "devplace.net",
                "origin": "https://devplace.net",
                "cookie": "session=abc",
            }
        )
    )
    assert headers["X-Forwarded-Host"] == "devplace.net"
    assert headers["origin"] == "https://devplace.net"
    assert headers["cookie"] == "session=abc"
    assert not [name for name in headers if name.lower() == "host"]


def test_ws_headers_drop_only_the_handshake_headers_the_client_regenerates():
    headers = forward.ws_headers(
        _proxy_websocket(
            {
                "host": "devplace.net",
                "x-real-ip": "203.0.113.7",
                "accept-language": "nl-NL",
                "sec-websocket-key": "should-not-survive",
                "sec-websocket-version": "13",
                "sec-websocket-extensions": "permessage-deflate",
                "sec-websocket-protocol": "v1",
            }
        )
    )
    assert headers["x-real-ip"] == "203.0.113.7"
    assert headers["accept-language"] == "nl-NL"
    for name in forward.WS_HANDSHAKE_HEADERS:
        assert name not in headers


def test_ws_headers_derive_the_forwarded_proto_from_the_socket_scheme():
    assert forward.ws_headers(_proxy_websocket({"host": "d.net"}))[
        "X-Forwarded-Proto"
    ] == "https"
    assert forward.ws_headers(_proxy_websocket({"host": "d.net"}, scheme="ws"))[
        "X-Forwarded-Proto"
    ] == "http"
    assert forward.ws_headers(
        _proxy_websocket({"host": "d.net", "x-forwarded-proto": "https"}, scheme="ws")
    )["X-Forwarded-Proto"] == "https"


def test_forward_headers_send_the_public_host_and_the_prefix():
    headers = forward.forward_headers(
        _proxy_request({"host": "devplace.net", "x-real-ip": "203.0.113.7"}),
        prefix="/p/demo",
    )
    assert headers["Host"] == "devplace.net"
    assert headers["X-Forwarded-Host"] == "devplace.net"
    assert headers["X-Forwarded-Prefix"] == "/p/demo"
    assert headers["X-Script-Name"] == "/p/demo"
    assert headers["x-real-ip"] == "203.0.113.7"


def test_ws_subprotocols_are_parsed_for_negotiation():
    assert forward.ws_subprotocols(_proxy_websocket({"host": "d.net"})) is None
    assert forward.ws_subprotocols(
        _proxy_websocket({"host": "d.net", "sec-websocket-protocol": "v2, v1"})
    ) == ["v2", "v1"]


def test_upstream_url_encodes_the_path_and_keeps_the_query_verbatim():
    assert (
        forward.upstream_url("http", "h:1", "dir/a b#c", "keep=1")
        == "http://h:1/dir/a%20b%23c?keep=1"
    )
    assert forward.upstream_url("ws", "h:1", "", "") == "ws://h:1/"


def test_raw_query_survives_a_hash_in_the_path():
    request = _proxy_request({"host": "d.net"}, query="keep=1")
    request.scope["path"] = "/weird/a b#c"
    assert forward.raw_query(request) == "keep=1"
    assert request.url.query == ""


def test_tunnel_target_prefers_the_direct_container_leg():
    instance = {
        "ports_json": '[{"host": 20500, "container": 8443, "proto": "tcp"}]',
        "container_gateway": "172.17.0.1",
        "container_ip": "172.17.0.9",
    }
    assert api.tunnel_target(instance, 8443) == ("172.17.0.9", 8443)


def test_tunnel_target_falls_back_to_the_published_port_without_a_container_ip():
    instance = {
        "ports_json": '[{"host": 20500, "container": 8443, "proto": "tcp"}]',
        "container_gateway": "172.17.0.1",
    }
    assert api.tunnel_target(instance, 8443) == ("172.17.0.1", 20500)


def test_proxy_target_and_tunnel_target_agree_on_the_same_port():
    instance = {
        "ports_json": '[{"host": 20500, "container": 8443, "proto": "tcp"}]',
        "container_gateway": "172.17.0.1",
        "container_ip": "172.17.0.9",
        "ingress_port": 8443,
    }
    assert api.proxy_target(instance) == api.tunnel_target(instance, 8443)


def test_tunnel_target_dials_the_container_for_an_unpublished_port():
    instance = {
        "ports_json": '[{"host": 20500, "container": 8443, "proto": "tcp"}]',
        "container_gateway": "172.17.0.1",
        "container_ip": "172.17.0.9",
    }
    assert api.tunnel_target(instance, 3000) == ("172.17.0.9", 3000)


def test_tunnel_target_is_empty_without_a_route_to_the_port():
    instance = {"ports_json": "[]", "container_gateway": "172.17.0.1"}
    assert api.tunnel_target(instance, 3000) == (None, None)
    assert api.tunnel_target({"container_ip": "172.17.0.9"}, 0) == (None, None)