windmill/backend/windmill-api-integration-tests/tests/datatable_roles.rs

//! Who may connect to a data table as which role, across the two shapes an entry can take: one
//! that owns its database, and a fork's pointer at it.

use serde_json::{json, Value};
use sqlx::{Pool, Postgres};

use windmill_test_utils::*;

fn client() -> reqwest::Client {
    reqwest::Client::new()
}

fn authed(builder: reqwest::RequestBuilder, token: &str) -> reqwest::RequestBuilder {
    builder.header("Authorization", format!("Bearer {token}"))
}

/// The `analytics` role's tenant list as stored, so a cascade can be observed directly.
async fn tenants(db: &Pool<Postgres>, w_id: &str) -> Vec<String> {
    let value: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main'->'permissions'->'roles'->'role1'->'tenants'
         FROM workspace_settings WHERE workspace_id = $1",
    )
    .bind(w_id)
    .fetch_one(db)
    .await
    .unwrap();
    serde_json::from_value(value.unwrap_or(json!([]))).unwrap()
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn freeing_a_principal_takes_its_datatable_tenant(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let base = format!("http://localhost:{port}/api/w/test-workspace");

    assert_eq!(
        tenants(&db, "test-workspace").await,
        vec!["u/test-user-2", "g/analysts", "f/finance"]
    );

    let resp = authed(
        client().delete(format!("{base}/groups/delete/analysts")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "delete group: {}", resp.text().await?);

    let resp = authed(
        client().delete(format!("{base}/folders/delete/finance")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "delete folder: {}", resp.text().await?);

    let resp = authed(
        client().delete(format!("{base}/users/delete/test-user-2")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "delete user: {}", resp.text().await?);

    // Leaving is the other way a membership ends, and there are two `/leave` routes — the one the
    // UI and the generated client call is this one. A tenant left behind here comes back with the
    // person on rejoin, or attaches to whoever takes the username next.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable,
             '{datatables,main,permissions,roles,role1,tenants}', '["u/test-user-3"]'::jsonb)
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;
    let resp = authed(
        client().post(format!("{base}/workspaces/leave")),
        "SECRET_TOKEN_3",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "leave: {}", resp.text().await?);
    // Nothing left naming a principal that no longer exists: a later group or account reusing one
    // of those names must not inherit the access this one had.
    assert!(
        tenants(&db, "test-workspace").await.is_empty(),
        "leaving kept the tenant: {:?}",
        tenants(&db, "test-workspace").await
    );
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_fork_uses_the_data_table_it_points_at_but_never_administers_it(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let fork = format!("http://localhost:{port}/api/w/wm-fork-dt/workspaces");

    // `test-user-2` is an admin of the fork and a plain member of the parent. The roles they can
    // use are the ones the parent's tenants give them there, not what their fork admin bit says.
    let resp = authed(
        client().get(format!("{fork}/datatable_usable_roles/main")),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await?;
    assert_eq!(body["roles"], json!(["analytics"]), "{body}");
    assert_eq!(body["default_role"], "analytics");

    // The drawer names the workspace that decides, and refuses to let the fork edit it.
    let resp = authed(
        client().get(format!("{fork}/datatable_permissions/main")),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(body["governing_workspace_id"], "test-workspace");
    assert_eq!(body["editable"], false, "{body}");

    let resp = authed(
        client().post(format!("{fork}/datatable_permissions/main")),
        "SECRET_TOKEN_2",
    )
    .json(&json!({"permissioned": true, "default_role": "admin",
                  "roles": [{"id": "admin", "tenants": ["*"]}]}))
    .send()
    .await?;
    assert_eq!(
        resp.status(),
        401,
        "a fork admin widened the parent's access"
    );

    // Nor by saving the settings form: the pointer is server-owned, so a payload naming the
    // parent's database leaves the entry exactly as it was.
    let resp = authed(
        client().post(format!("{fork}/edit_datatable_config")),
        "SECRET_TOKEN_2",
    )
    .json(&json!({
        "settings": {"datatables": {"main": {
            "database": {"resource_type": "instance", "resource_path": "dt_main"}
        }}},
        "renames": [], "deleted_datatables": []
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main' FROM workspace_settings WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    let entry = entry.unwrap();
    assert_eq!(
        entry["reference"]["workspace_id"], "test-workspace",
        "{entry}"
    );
    assert!(entry["database"].is_null(), "{entry}");
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_second_entry_on_the_same_database_is_reported_rather_than_governed(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A copy of the parent's entry, as a fork created before data table roles would hold. It keeps
    // its own access, so the owner is told about it instead of being told it is covered.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = '{"datatables": {"copy": {
            "database": {"resource_type": "instance", "resource_path": "dt_main"}}}}'::jsonb
           WHERE workspace_id = 'wm-fork-dt'"#,
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let resp = authed(
        client().get(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/datatable_permissions/main"
        )),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(
        body["ungoverned_reachers"],
        json!([{"workspace_id": "wm-fork-dt", "datatable": "copy"}]),
        "{body}"
    );
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_resource_backed_data_table_cannot_be_put_under_roles(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A role is a login on Windmill's own cluster. A resource-backed data table dials a host the
    // workspace admin chose, so accepting one here would hand that host a real cluster credential.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = '{"datatables": {"byo": {
            "database": {"resource_type": "postgresql", "resource_path": "u/test-user/pg"}}}}'::jsonb
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let base = format!("http://localhost:{port}/api/w/test-workspace/workspaces");

    let resp = authed(
        client().get(format!("{base}/datatable_permissions/byo")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(body["supported"], false, "{body}");

    let resp = authed(
        client().post(format!("{base}/datatable_permissions/byo")),
        "SECRET_TOKEN",
    )
    .json(&json!({"permissioned": true, "default_role": "role1",
                  "roles": [{"id": "role1", "tenants": ["*"]}]}))
    .send()
    .await?;
    assert_eq!(resp.status(), 400, "{}", resp.text().await?);
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_fork_renaming_its_own_entry_leaves_the_governing_bookkeeping_alone(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // The parent's migration definitions. A rename or delete through the fork's settings form
    // resolves through the pointer, so without a guard it would relabel or wipe these.
    sqlx::query(
        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up)
         VALUES ('test-workspace', 'main', 1, 'init', 'SELECT 1')",
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/wm-fork-dt/workspaces/edit_datatable_config"
        )),
        "SECRET_TOKEN_2",
    )
    .json(&json!({
        "settings": {"datatables": {}},
        "renames": [],
        "deleted_datatables": ["main"]
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let left: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM datatable_migrations WHERE workspace_id = 'test-workspace'",
    )
    .fetch_one(&db)
    .await?;
    assert_eq!(left, 1, "the fork's delete reached the parent's migrations");
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_caller_who_is_not_a_member_of_the_governing_workspace_reaches_nothing(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A fork member who was never added to the parent. Their fork membership says nothing there,
    // and the email lookup that would evaluate them as a member of it finds no row.
    sqlx::query(
        "INSERT INTO usr (workspace_id, email, username, is_admin, role)
         VALUES ('wm-fork-dt', 'test3@windmill.dev', 'test-user-3', false, 'User')",
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let resp = authed(
        client().get(format!(
            "http://localhost:{port}/api/w/wm-fork-dt/workspaces/datatable_usable_roles/main"
        )),
        "SECRET_TOKEN_3",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await?;
    assert_eq!(body["roles"], json!([]), "{body}");
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_caller_with_no_identity_reaches_a_permissioned_data_table_not_at_all(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    use windmill_common::workspaces::{get_datatable_resource_from_db, DatatableAccess};

    initialize_tracing().await;
    // The compatibility story for an agent worker that predates data table roles and sends no job
    // id: it keeps resolving an unpermissioned data table, and is refused on a permissioned one
    // rather than handed an unattributed admin connection.
    let refused = get_datatable_resource_from_db(
        &db,
        "test-workspace",
        "main",
        None,
        DatatableAccess::NoIdentity,
    )
    .await;
    assert!(refused.is_err(), "an unidentified caller was let in");

    sqlx::query(
        "UPDATE workspace_settings
         SET datatable = datatable #- '{datatables,main,permissions}'
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;
    let resolved = get_datatable_resource_from_db(
        &db,
        "test-workspace",
        "main",
        None,
        DatatableAccess::NoIdentity,
    )
    .await?;
    assert_eq!(resolved["dbname"], "dt_main", "{resolved}");
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn concurrent_role_creations_both_survive(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Postgres roles are cluster-wide and this cluster is shared with every other test database,
    // so the names have to be unique to this run.
    let suffix: String = uuid::Uuid::new_v4().simple().to_string()[..8].to_string();
    let names = [format!("wmtest_a_{suffix}"), format!("wmtest_b_{suffix}")];

    // The cluster DDL is not visible to another transaction until commit, so without the lock both
    // of these pass their `pg_roles` existence check and one loses — leaving a live cluster login
    // the catalog never recorded.
    let create = |name: String| async move {
        let resp = authed(
            client().post(format!(
                "http://localhost:{port}/api/settings/datatable_roles"
            )),
            "SECRET_TOKEN",
        )
        .json(&json!({ "name": name }))
        .send()
        .await?;
        let status = resp.status();
        let body = resp.text().await?;
        Ok::<_, anyhow::Error>((status, body))
    };
    let outcome = async {
        let (a, b) = tokio::join!(create(names[0].clone()), create(names[1].clone()));
        let (a, b) = (a?, b?);
        assert_eq!(a.0, 200, "{}", a.1);
        assert_eq!(b.0, 200, "{}", b.1);

        let catalog = windmill_common::datatable_roles::read_role_catalog(
            &db,
            windmill_common::datatable_roles::DatatableRoleCluster::Instance,
        )
        .await?;
        let recorded: Vec<&str> = catalog.values().map(|r| r.name.as_str()).collect();
        for name in &names {
            assert!(
                recorded.contains(&name.as_str()),
                "{name} is a live cluster login the catalog forgot: {recorded:?}"
            );
        }
        Ok::<_, anyhow::Error>(())
    }
    .await;

    // Roles are cluster-wide, so they outlive this test's throwaway database. Dropped whatever
    // happened above — a failing run is exactly the one that created them and did not record them.
    for name in &names {
        let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{name}\""))
            .execute(&db)
            .await;
    }
    outcome
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn an_existing_role_is_taken_over_only_when_asked_and_never_with_members(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    let suffix: String = uuid::Uuid::new_v4().simple().to_string()[..8].to_string();
    let plain = format!("wmtest_plain_{suffix}");
    let group = format!("wmtest_group_{suffix}");
    let member = format!("wmtest_member_{suffix}");
    for sql in [
        // An expired password and no connection slot: kept as is, the login Windmill sets is dead.
        format!(
            "CREATE ROLE \"{plain}\" LOGIN PASSWORD 'theirs' VALID UNTIL '2000-01-01' \
             CONNECTION LIMIT 0"
        ),
        format!("CREATE ROLE \"{group}\" NOLOGIN"),
        format!("CREATE ROLE \"{member}\" LOGIN"),
        format!("GRANT \"{group}\" TO \"{member}\""),
    ] {
        sqlx::query(&sql).execute(&db).await?;
    }

    let create = |name: String, take_over: bool| async move {
        let resp = authed(
            client().post(format!(
                "http://localhost:{port}/api/settings/datatable_roles"
            )),
            "SECRET_TOKEN",
        )
        .json(&json!({ "name": name, "take_over": take_over }))
        .send()
        .await?;
        Ok::<_, anyhow::Error>((resp.status().as_u16(), resp.text().await?))
    };
    let outcome = async {
        let (status, body) = create(plain.clone(), false).await?;
        assert_eq!(
            status, 409,
            "an existing role is not taken without asking: {body}"
        );

        // The group's member would inherit every grant the data table role gets.
        let (status, body) = create(group.clone(), true).await?;
        assert_eq!(status, 400, "{body}");
        assert!(
            body.contains(&member),
            "the refusal names the member: {body}"
        );

        let (status, body) = create(plain.clone(), true).await?;
        assert_eq!(status, 200, "{body}");
        let catalog = windmill_common::datatable_roles::read_role_catalog(
            &db,
            windmill_common::datatable_roles::DatatableRoleCluster::Instance,
        )
        .await?;
        let taken = catalog
            .values()
            .find(|r| r.name == plain)
            .ok_or_else(|| anyhow::anyhow!("{plain} is not in the catalog: {catalog:?}"))?;
        let options = (*db.connect_options())
            .clone()
            .username(&plain)
            .password(taken.pwd.as_deref().unwrap_or_default());
        let mut login = <sqlx::PgConnection as sqlx::Connection>::connect_with(&options).await?;
        sqlx::query("SELECT 1").execute(&mut login).await?;
        Ok::<_, anyhow::Error>(())
    }
    .await;

    // Roles are cluster-wide and outlive this test's database, whatever happened above.
    for name in [&member, &group, &plain] {
        let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{name}\""))
            .execute(&db)
            .await;
    }
    outcome
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_role_delete_that_fails_part_way_leaves_the_role_disabled(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let suffix: String = uuid::Uuid::new_v4().simple().to_string()[..8].to_string();
    let name = format!("wmtest_del_{suffix}");

    let outcome = async {
        let created: Value = authed(
            client().post(format!(
                "http://localhost:{port}/api/settings/datatable_roles"
            )),
            "SECRET_TOKEN",
        )
        .json(&json!({ "name": name }))
        .send()
        .await?
        .error_for_status()?
        .json()
        .await?;
        let id = created["id"].as_str().unwrap().to_string();

        // Each database's pass commits on its own, so one that cannot be reached fails the delete
        // after the others may already have stripped the role.
        sqlx::query(
            "UPDATE global_settings SET value = jsonb_set(value, '{databases,wm_unreachable}', '{}')
             WHERE name = 'custom_instance_pg_databases'",
        )
        .execute(&db)
        .await?;

        let resp = authed(
            client().delete(format!(
                "http://localhost:{port}/api/settings/datatable_roles/{id}"
            )),
            "SECRET_TOKEN",
        )
        .send()
        .await?;
        let status = resp.status();
        let body = resp.text().await?;
        assert_eq!(status, 400, "{body}");

        let catalog = windmill_common::datatable_roles::read_role_catalog(
            &db,
            windmill_common::datatable_roles::DatatableRoleCluster::Instance,
        )
        .await?;
        let role = catalog
            .get(&id)
            .expect("a failed delete keeps the entry to retry");
        assert!(
            !role.enabled,
            "a half-deleted role is still enabled in the catalog"
        );
        let can_login: bool =
            sqlx::query_scalar("SELECT rolcanlogin FROM pg_roles WHERE rolname = $1")
                .bind(&name)
                .fetch_one(&db)
                .await?;
        assert!(!can_login, "a half-deleted role can still log in");
        Ok::<_, anyhow::Error>(())
    }
    .await;

    let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{name}\""))
        .execute(&db)
        .await;
    outcome
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn renaming_a_governing_data_table_carries_its_forks(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // A pointer names the governing data table by name, so a rename that does not follow leaves
    // every fork resolving to nothing — the data table vanishes from their pickers and their jobs
    // stop, with nothing in the renaming workspace to suggest why.
    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/edit_datatable_config"
        )),
        "SECRET_TOKEN",
    )
    .json(&json!({
        "settings": {"datatables": {"renamed": {
            "database": {"resource_type": "instance", "resource_path": "dt_main"}
        }}},
        "renames": [{"from": "main", "to": "renamed"}],
        "deleted_datatables": []
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main' FROM workspace_settings WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    let entry = entry.unwrap();
    assert_eq!(entry["reference"]["datatable"], "renamed", "{entry}");

    // And it still resolves, which is the thing the fork actually cares about.
    let resp = authed(
        client().get(format!(
            "http://localhost:{port}/api/w/wm-fork-dt/workspaces/datatable_usable_roles/main"
        )),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_rename_has_to_match_the_save_it_claims_to_describe(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let url =
        format!("http://localhost:{port}/api/w/test-workspace/workspaces/edit_datatable_config");

    let instance =
        |path: &str| json!({"database": {"resource_type": "instance", "resource_path": path}});

    // Fork pointers are rewritten from the rename list, so a rename nobody performed moves every
    // fork of one data table onto another. `main` survives this save, so it was not renamed.
    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&json!({
            "settings": {"datatables": {"main": instance("dt_main"), "decoy": instance("dt_two")}},
            "renames": [{"from": "main", "to": "decoy"}],
            "deleted_datatables": []
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 400, "a forged rename was accepted");

    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main'->'reference' FROM workspace_settings
         WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    assert_eq!(
        entry.unwrap()["datatable"],
        "main",
        "the fork was repointed anyway"
    );

    // A swap is two renames whose sources and targets cross. It cannot be done one at a time —
    // `datatables` is keyed by name — so refusing it would be a regression, and applying the two
    // in order without a temporary name would carry `main`'s pointers back to `main`.
    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&json!({
            "settings": {"datatables": {"main": instance("dt_two"), "other": instance("dt_main")}},
            "renames": [{"from": "main", "to": "other"}, {"from": "other", "to": "main"}],
            "deleted_datatables": []
        }))
        .send()
        .await?;
    // `other` does not exist yet, so this particular pair is still refused — the swap shape is
    // covered by the pair below, which starts from two real data tables.
    assert_eq!(resp.status(), 400, "{}", resp.text().await?);

    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,other}',
             '{"database": {"resource_type": "instance", "resource_path": "dt_two"}}'::jsonb)
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;

    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&json!({
            "settings": {"datatables": {"main": instance("dt_two"), "other": instance("dt_main")}},
            "renames": [{"from": "main", "to": "other"}, {"from": "other", "to": "main"}],
            "deleted_datatables": []
        }))
        .send()
        .await?;
    assert_eq!(
        resp.status(),
        200,
        "a swap was refused: {}",
        resp.text().await?
    );

    // The fork named `main`, which is now called `other`.
    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main'->'reference' FROM workspace_settings
         WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    assert_eq!(
        entry.unwrap()["datatable"],
        "other",
        "the swap did not carry the pointer"
    );
    Ok(())
}

async fn database_exists(db: &Pool<Postgres>, name: &str) -> anyhow::Result<bool> {
    Ok(
        sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = $1)")
            .bind(name)
            .fetch_one(db)
            .await?,
    )
}

async fn workspace_exists(db: &Pool<Postgres>, id: &str) -> anyhow::Result<bool> {
    Ok(
        sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM workspace WHERE id = $1)")
            .bind(id)
            .fetch_one(db)
            .await?,
    )
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_data_table_under_roles_is_cloned_only_with_its_grants(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    // The database a fork's copy of `main` goes into is named after the fork. Every request below
    // is refused before it is created, so the cluster-wide name never collides with a sibling run.
    let target = "wm_fork_copy__main";
    let fork = |token: &str, forked: Value| {
        authed(
            client().post(format!(
                "http://localhost:{port}/api/w/test-workspace/workspaces/create_fork"
            )),
            token,
        )
        .json(&json!({"id": "wm-fork-copy", "name": "copy", "forked_datatables": [forked]}))
    };

    // Rows are a workspace admin's to copy, as they were before roles.
    let resp = fork(
        "SECRET_TOKEN_2",
        json!({"name": "main", "new_dbname": target, "fork_behavior": "schema_and_data"}),
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 403, "{}", resp.text().await?);
    assert!(!database_exists(&db, target).await?);

    // Even the schema alone lists every table, which a member covered by no role cannot read in
    // the parent.
    let resp = fork(
        "SECRET_TOKEN_3",
        json!({"name": "main", "new_dbname": target, "fork_behavior": "schema_only"}),
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 401, "{}", resp.text().await?);
    assert!(!database_exists(&db, target).await?);

    // Refused in the first phase too, before a git branch is created for a fork that cannot be.
    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/create_workspace_fork_branch"
        )),
        "SECRET_TOKEN_3",
    )
    .json(
        &json!({"id": "wm-fork-copy", "name": "copy", "forked_datatables": [
            {"name": "main", "new_dbname": target, "fork_behavior": "schema_only"}
        ]}),
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 401, "{}", resp.text().await?);

    // A database the request did not create — whatever the data table, under roles or not — may be
    // a copy left behind by a deleted fork, which the entry would reach without its governance.
    let resp = fork(
        "SECRET_TOKEN",
        json!({"name": "other", "new_dbname": "wm_fork_copy__other"}),
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 400);
    assert!(
        resp.text().await?.contains("fork_behavior"),
        "a database the fork did not create was taken"
    );
    assert!(!workspace_exists(&db, "wm-fork-copy").await?);

    // The copy an admin asks for goes ahead in an edition that replays grants, and is refused
    // before any database exists in one that does not. The fixture's database does not exist, so
    // the dump fails either way — and leaves neither a database nor a fork behind.
    let resp = fork(
        "SECRET_TOKEN",
        json!({"name": "main", "new_dbname": target, "fork_behavior": "schema_only"}),
    )
    .send()
    .await?;
    assert_ne!(resp.status(), 200);
    let body = resp.text().await?;
    #[cfg(all(feature = "private", feature = "enterprise"))]
    assert!(body.contains("pg_dump"), "{body}");
    #[cfg(not(all(feature = "private", feature = "enterprise")))]
    assert!(body.contains("Enterprise Edition"), "{body}");
    assert!(!database_exists(&db, target).await?);
    assert!(!workspace_exists(&db, "wm-fork-copy").await?);
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_clone_takes_its_roles_from_the_data_table_it_was_cloned_from(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,copy}', '{
               "database": {"resource_type": "instance", "resource_path": "wm_fork_dt__copy"},
               "governed_by": {"workspace_id": "test-workspace", "datatable": "main"},
               "forked_from": {}
           }'::jsonb) WHERE workspace_id = 'wm-fork-dt'"#,
    )
    .execute(&db)
    .await?;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let fork = format!("http://localhost:{port}/api/w/wm-fork-dt/workspaces");

    // `test-user-2` administers the fork, and is only a tenant of `analytics` in the parent.
    let resp = authed(
        client().get(format!("{fork}/datatable_usable_roles/copy")),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(body["roles"], json!(["analytics"]), "{body}");

    let resp = authed(
        client().get(format!("{fork}/datatable_permissions/copy")),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(body["editable"], false, "{body}");
    assert_eq!(body["clone_of"]["workspace_id"], "test-workspace", "{body}");

    // Nobody changes a clone's roles, a superadmin included: they are the source's.
    for token in ["SECRET_TOKEN_2", "SECRET_TOKEN"] {
        let resp = authed(
            client().post(format!("{fork}/datatable_permissions/copy")),
            token,
        )
        .json(&json!({"permissioned": false}))
        .send()
        .await?;
        assert_eq!(resp.status(), 400, "{token} changed a clone's roles");
    }

    // A settings save cannot clear the link.
    let resp = authed(
        client().post(format!("{fork}/edit_datatable_config")),
        "SECRET_TOKEN_2",
    )
    .json(&json!({
        "settings": {"datatables": {"main": {}, "copy": {
            "database": {"resource_type": "instance", "resource_path": "wm_fork_dt__copy"}
        }}},
        "renames": [], "deleted_datatables": []
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    let governed_by: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'copy'->'governed_by' FROM workspace_settings
         WHERE workspace_id = 'wm-fork-dt'",
    )
    .fetch_one(&db)
    .await?;
    assert_eq!(governed_by.unwrap()["datatable"], "main");

    // Nor move it, a superadmin included: its grants were replayed into that database alone.
    let resp = authed(
        client().post(format!("{fork}/edit_datatable_config")),
        "SECRET_TOKEN",
    )
    .json(&json!({
        "settings": {"datatables": {"main": {}, "copy": {
            "database": {"resource_type": "instance", "resource_path": "wm_fork_dt__elsewhere"}
        }}},
        "renames": [], "deleted_datatables": []
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 400, "{}", resp.text().await?);

    // Nor reach its database through a second entry without roles, which would connect as `admin`.
    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/edit_datatable_config"
        )),
        "SECRET_TOKEN",
    )
    .json(&json!({
        "settings": {"datatables": {
            "main": {"database": {"resource_type": "instance", "resource_path": "dt_main"}},
            "alias": {"database": {"resource_type": "instance", "resource_path": "wm_fork_dt__copy"}}
        }},
        "renames": [], "deleted_datatables": []
    }))
    .send()
    .await?;
    let status = resp.status();
    let body = resp.text().await?;
    assert_eq!(status, 400, "{body}");
    assert!(body.contains("without carrying those roles"), "{body}");
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_data_table_under_roles_is_not_copied_without_its_grants(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/create_pg_database"
        )),
        "SECRET_TOKEN",
    )
    .json(&json!({"source": "datatable://main", "target_dbname": "wm_fork_dt_copy"}))
    .send()
    .await?;
    assert_eq!(resp.status(), 400);
    assert!(
        resp.text().await?.contains("under roles"),
        "the fork's database was created for a copy that cannot happen"
    );

    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/import_pg_database"
        )),
        "SECRET_TOKEN",
    )
    .json(
        &json!({"source": "datatable://main", "target": "datatable://main",
                  "fork_behavior": "schema_only"}),
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 400);
    assert!(
        resp.text().await?.contains("under roles"),
        "the copy was refused for some other reason"
    );
    Ok(())
}

/// The fork's `forked_from` for one of its entries; `None` whether it is absent or `null`.
async fn forked_from_of(db: &Pool<Postgres>, name: &str) -> Option<Value> {
    sqlx::query_scalar::<_, Option<Value>>(
        "SELECT datatable->'datatables'->$1::text->'forked_from'
         FROM workspace_settings WHERE workspace_id = 'wm-fork-dt'",
    )
    .bind(name)
    .fetch_one(db)
    .await
    .unwrap()
    .filter(|v| !v.is_null())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_clone_stamp_is_carried_but_its_schema_baseline_advances(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // Whether an entry is a clone is what marks its database droppable, so a save can neither
    // stamp nor unstamp one. The schema baseline inside the stamp is what the fork's schema diff
    // advances after applying a change; dropping it would offer that same change again.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = '{"datatables": {
             "clone": {"database": {"resource_type": "instance", "resource_path": "wm_fork_dt__clone"},
                       "forked_from": {"schema": {}}},
             "plain": {"database": {"resource_type": "instance", "resource_path": "dt_plain"}}}}'::jsonb
           WHERE workspace_id = 'wm-fork-dt'"#,
    )
    .execute(&db)
    .await?;
    let server = ApiServer::start(db.clone()).await?;
    let url = format!(
        "http://localhost:{}/api/w/wm-fork-dt/workspaces/edit_datatable_config",
        server.addr.port()
    );
    let clone_db = json!({"resource_type": "instance", "resource_path": "wm_fork_dt__clone"});
    let plain_db = json!({"resource_type": "instance", "resource_path": "dt_plain"});
    let baseline = json!({"schema": {"public": {"orders": {"id": "int4"}}}});

    let resp = authed(client().post(&url), "SECRET_TOKEN_2")
        .json(&json!({"settings": {"datatables": {
            "clone": {"database": clone_db, "forked_from": baseline},
            "plain": {"database": plain_db, "forked_from": {"schema": {}}}
        }}}))
        .send()
        .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    assert_eq!(
        forked_from_of(&db, "clone").await,
        Some(baseline.clone()),
        "the schema diff's baseline did not advance"
    );
    assert_eq!(
        forked_from_of(&db, "plain").await,
        None,
        "a save stamped a clone"
    );

    let resp = authed(client().post(&url), "SECRET_TOKEN_2")
        .json(&json!({"settings": {"datatables": {
            "clone": {"database": clone_db}, "plain": {"database": plain_db}
        }}}))
        .send()
        .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    assert_eq!(
        forked_from_of(&db, "clone").await,
        Some(baseline),
        "a save unstamped a clone"
    );
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn roles_cannot_be_turned_on_while_a_trigger_streams_the_data_table(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A replication stream reads every row whatever the roles grant, so a data table carries one
    // or the other. An enabled trigger on it — here a fork's, through its pointer — keeps roles
    // from being turned on, and disabling it is what lets them on.
    sqlx::query(
        "UPDATE workspace_settings SET datatable = datatable #- '{datatables,main,permissions}'
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;
    sqlx::query(
        r#"INSERT INTO postgres_trigger (path, script_path, is_flow, workspace_id, edited_by,
             postgres_resource_path, replication_slot_name, publication_name, permissioned_as, mode)
           VALUES ('u/test-user-2/fork_stream', 'u/test-user-2/s', false, 'wm-fork-dt',
                   'test-user-2', 'datatable://main', 'slot_fork', 'pub_fork', 'u/test-user-2',
                   'enabled')"#,
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let url = format!(
        "http://localhost:{}/api/w/test-workspace/workspaces/datatable_permissions/main",
        server.addr.port()
    );
    let turn_on = json!({"permissioned": true, "default_role": "admin",
                         "roles": [{"id": "admin", "tenants": ["*"]}]});

    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&turn_on)
        .send()
        .await?;
    assert_eq!(resp.status(), 400);
    assert!(
        resp.text()
            .await?
            .contains("wm-fork-dt/u/test-user-2/fork_stream"),
        "the refusal does not name the trigger to disable"
    );

    // Disabled, but its listener pinged just now and stops only at its next heartbeat.
    sqlx::query(
        "UPDATE postgres_trigger SET mode = 'disabled', server_id = NULL, last_server_ping = now()
         WHERE path = 'u/test-user-2/fork_stream'",
    )
    .execute(&db)
    .await?;
    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&turn_on)
        .send()
        .await?;
    assert_eq!(
        resp.status(),
        400,
        "roles went on while a disabled trigger's listener was still attached: {}",
        resp.text().await?
    );

    sqlx::query(
        "UPDATE postgres_trigger SET last_server_ping = now() - interval '20 seconds'
         WHERE path = 'u/test-user-2/fork_stream'",
    )
    .execute(&db)
    .await?;
    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&turn_on)
        .send()
        .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn roles_going_on_wait_for_a_trigger_being_enabled(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    sqlx::query(
        "UPDATE workspace_settings SET datatable = datatable #- '{datatables,main,permissions}'
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;

    // A trigger enable in flight: it holds the stream lock and its row is not committed yet, so a
    // roles save that looked for streams now would miss it and its listener would connect to a
    // data table it is about to be refused.
    let mut enabling = db.begin().await?;
    windmill_common::datatable_roles::lock_datatable_streams(&mut *enabling, false).await?;
    sqlx::query(
        r#"INSERT INTO postgres_trigger (path, script_path, is_flow, workspace_id, edited_by,
             postgres_resource_path, replication_slot_name, publication_name, permissioned_as, mode)
           VALUES ('u/test-user-2/racing_stream', 'u/test-user-2/s', false, 'wm-fork-dt',
                   'test-user-2', 'datatable://main', 'slot_race', 'pub_race', 'u/test-user-2',
                   'enabled')"#,
    )
    .execute(&mut *enabling)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let url = format!(
        "http://localhost:{}/api/w/test-workspace/workspaces/datatable_permissions/main",
        server.addr.port()
    );
    let save = tokio::spawn(
        authed(client().post(&url), "SECRET_TOKEN")
            .json(&json!({"permissioned": true, "default_role": "admin",
                          "roles": [{"id": "admin", "tenants": ["*"]}]}))
            .send(),
    );
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    assert!(
        !save.is_finished(),
        "roles went on while a trigger was being enabled"
    );
    enabling.commit().await?;

    let resp = save.await??;
    assert_eq!(resp.status(), 400);
    assert!(
        resp.text()
            .await?
            .contains("wm-fork-dt/u/test-user-2/racing_stream"),
        "the roles save missed the trigger enabled while it waited"
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_stored_name_containing_a_question_mark_resolves_as_itself(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // Names could contain `?` before they were restricted, and such an entry is still stored.
    sqlx::query(
        "UPDATE workspace_settings
         SET datatable = jsonb_set(datatable, '{datatables,legacy?dt}', datatable->'datatables'->'main')
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;

    let resolve = |reference: &'static str| {
        let db = db.clone();
        async move {
            windmill_common::workspaces::parse_datatable_ref_for(&db, "test-workspace", reference)
                .await
        }
    };
    assert_eq!(resolve("legacy?dt").await?, ("legacy?dt".to_string(), None));
    assert_eq!(
        resolve("main?role=analytics").await?,
        ("main".to_string(), Some("analytics".to_string()))
    );
    assert!(
        resolve("main?dt").await.is_err(),
        "an unknown parameter was ignored"
    );

    sqlx::query(
        "UPDATE workspace_settings
         SET datatable = jsonb_set(datatable, '{datatables,main?role=analytics}', datatable->'datatables'->'main')
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;
    assert!(
        resolve("main?role=analytics").await.is_err(),
        "a reference naming both a stored data table and a role on another resolved to one of them"
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_settings_save_dropping_a_governing_entry_names_the_forks_it_strands(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    // The whole map and no `deleted_datatables`, as a settings sync sends it.
    let resp = authed(
        client().post(format!(
            "http://localhost:{}/api/w/test-workspace/workspaces/edit_datatable_config",
            server.addr.port()
        )),
        "SECRET_TOKEN",
    )
    .json(&json!({ "settings": { "datatables": {} } }))
    .send()
    .await?;
    let status = resp.status();
    let body = resp.text().await?;
    assert_eq!(status, 200, "{body}");
    let result: Value = serde_json::from_str(&body)?;
    assert!(
        result["stranded_references"]
            .as_array()
            .is_some_and(|refs| refs.iter().any(|r| r["workspace_id"] == "wm-fork-dt")),
        "the fork left pointing at nothing was not named: {body}"
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn an_entry_without_roles_cannot_newly_reach_a_database_under_roles(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,other}',
             '{"database": {"resource_type": "instance", "resource_path": "dt_other"}}')
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Whole-map saves with no `renames`, as a settings sync sends them.
    let dt_main =
        json!({ "database": { "resource_type": "instance", "resource_path": "dt_main" } });
    let dt_other =
        json!({ "database": { "resource_type": "instance", "resource_path": "dt_other" } });
    for (case, w_id, datatables) in [
        (
            "a rename to a new name",
            "test-workspace",
            json!({ "main_renamed": dt_main, "other": dt_other }),
        ),
        (
            "an existing name repointed",
            "test-workspace",
            json!({ "other": dt_main }),
        ),
        (
            "another workspace's entry",
            "wm-fork-dt",
            json!({ "direct": dt_main }),
        ),
    ] {
        let resp = authed(
            client().post(format!(
                "http://localhost:{port}/api/w/{w_id}/workspaces/edit_datatable_config"
            )),
            "SECRET_TOKEN",
        )
        .json(&json!({ "settings": { "datatables": datatables } }))
        .send()
        .await?;
        let status = resp.status();
        let body = resp.text().await?;
        assert!(
            status == 400 && body.contains("which a data table under roles uses"),
            "{case} reached the database under roles without them ({status}): {body}"
        );
    }

    let still_governed: bool = sqlx::query_scalar(
        "SELECT (datatable->'datatables'->'main') ? 'permissions' FROM workspace_settings
         WHERE workspace_id = 'test-workspace'",
    )
    .fetch_one(&db)
    .await?;
    assert!(still_governed, "the refused save still took effect");
    Ok(())
}

/// Browsing names the role it connects as, and a role the caller may not use is refused rather
/// than quietly listed as the default. The refusal is decided before connecting, so the fixture's
/// database never has to exist.
#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn browsing_as_a_role_the_caller_may_not_use_is_refused(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let base = format!("http://localhost:{port}/api/w/test-workspace/workspaces");

    // `test-user-2` is a tenant of `analytics` only.
    let resp = authed(
        client().get(format!(
            "{base}/list_datatable_tables?role_for=main&role=admin"
        )),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await?;
    let entry = body
        .as_array()
        .and_then(|a| a.iter().find(|e| e["datatable_name"] == "main"))
        .expect("main is listed");
    assert_eq!(entry["usable_roles"], json!(["analytics"]), "{entry}");
    assert_eq!(entry["default_role"], "analytics", "{entry}");
    assert_eq!(entry["permissioned"], true, "{entry}");
    assert_eq!(entry["instance"], true, "{entry}");
    let error = entry["error"].as_str().unwrap_or_default();
    assert!(
        error.contains("Not allowed to use role 'admin'"),
        "listed as another role than the one asked for: {entry}"
    );

    let resp = authed(
        client().get(format!(
            "{base}/get_datatable_table_schema?datatable_name=main&schema_name=public&table_name=t&role=admin"
        )),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    let status = resp.status();
    let text = resp.text().await?;
    assert!(
        text.contains("Not allowed to use role 'admin'"),
        "{status}: {text}"
    );

    // A role means nothing without the data table it belongs to.
    let resp = authed(
        client().get(format!("{base}/list_datatable_tables?role=analytics")),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 400, "{}", resp.text().await?);
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn an_alias_saved_elsewhere_waits_for_roles_going_on_for_its_database(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,other}',
             '{"database": {"resource_type": "instance", "resource_path": "dt_other"}}')
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;

    // Roles going on for `dt_other`, not committed yet: it holds only its own workspace's settings
    // row, so an alias saved from another workspace that looked for roles now would miss them.
    let enabling = {
        let mut tx = db.begin().await?;
        windmill_common::datatable_roles::lock_instance_databases_governance(
            &mut *tx,
            ["dt_other"],
        )
        .await?;
        sqlx::query(
            r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable,
                 '{datatables,other,permissions}',
                 '{"default_role": "admin", "roles": {"admin": {"tenants": ["*"]}}}')
               WHERE workspace_id = 'test-workspace'"#,
        )
        .execute(&mut *tx)
        .await?;
        tx
    };

    let server = ApiServer::start(db.clone()).await?;
    let url = format!(
        "http://localhost:{}/api/w/wm-fork-dt/workspaces/edit_datatable_config",
        server.addr.port()
    );
    let save = tokio::spawn(
        authed(client().post(&url), "SECRET_TOKEN")
            .json(&json!({ "settings": { "datatables": {
                "direct": { "database": { "resource_type": "instance", "resource_path": "dt_other" } }
            } } }))
            .send(),
    );
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    assert!(
        !save.is_finished(),
        "an alias was saved while roles were going on for its database"
    );
    enabling.commit().await?;

    let resp = save.await??;
    let status = resp.status();
    let body = resp.text().await?;
    assert!(
        status == 400 && body.contains("which a data table under roles uses"),
        "the alias reached the database whose roles went on while it waited ({status}): {body}"
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_fork_waits_for_a_rename_of_the_data_table_it_keeps(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A rename in flight holds the parent's settings row and moves only the pointers it can see; a
    // fork being created is invisible to it, so the fork has to read the name the rename commits.
    let mut renaming = db.begin().await?;
    sqlx::query(
        "SELECT 1 FROM workspace_settings WHERE workspace_id = 'test-workspace' FOR UPDATE",
    )
    .execute(&mut *renaming)
    .await?;
    sqlx::query(
        "UPDATE workspace_settings SET datatable = jsonb_set(datatable #- '{datatables,main}',
           '{datatables,renamed}', datatable->'datatables'->'main')
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&mut *renaming)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let url = format!(
        "http://localhost:{}/api/w/test-workspace/workspaces/create_fork",
        server.addr.port()
    );
    let fork = tokio::spawn(
        authed(client().post(&url), "SECRET_TOKEN")
            .json(&json!({ "id": "wm-fork-race", "name": "race", "color": "#0000ff" }))
            .send(),
    );
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    assert!(
        !fork.is_finished(),
        "the fork copied the parent's data tables while a rename held them"
    );
    renaming.commit().await?;

    let resp = fork.await??;
    assert!(resp.status().is_success(), "{}", resp.text().await?);
    let datatables: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables' FROM workspace_settings WHERE workspace_id = 'wm-fork-race'",
    )
    .fetch_one(&db)
    .await?;
    let datatables = datatables.unwrap();
    assert_eq!(
        datatables["renamed"]["reference"],
        json!({ "workspace_id": "test-workspace", "datatable": "renamed" }),
        "{datatables}"
    );
    Ok(())
}

async fn enable_migrations_on_main(db: &Pool<Postgres>) -> anyhow::Result<()> {
    sqlx::query(
        "UPDATE workspace_settings
         SET datatable = jsonb_set(datatable, '{datatables,main,migrations_enabled}', 'true')
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(db)
    .await?;
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_migration_is_written_only_by_who_may_run_it(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    enable_migrations_on_main(&db).await?;
    let server = ApiServer::start(db.clone()).await?;
    let base = format!(
        "http://localhost:{}/api/w/test-workspace/workspaces",
        server.addr.port()
    );
    let create = |token: &str, body: Value| {
        authed(
            client()
                .post(format!("{base}/create_datatable_migration/main"))
                .json(&body),
            token,
        )
        .send()
    };
    let upsert = |token: &str, code_up: &str| {
        authed(
            client()
                .post(format!("{base}/upsert_datatable_migration/main"))
                .json(&json!({"timestamp": 20260101000000i64, "name": "m", "code_up": code_up})),
            token,
        )
        .send()
    };

    // `test-user-2` holds `analytics` and nothing else, so of a migration they may store only
    // what they could run: SQL that declares no role runs as admin, in the up or the down.
    for body in [
        json!({"name": "m", "code_up": "DROP TABLE t"}),
        json!({"name": "m", "code_up": "-- role analytics\nSELECT 1", "code_down": "DROP TABLE t"}),
    ] {
        let resp = create("SECRET_TOKEN_2", body.clone()).await?;
        assert_eq!(resp.status(), 401, "{body}: {}", resp.text().await?);
    }
    let resp = create(
        "SECRET_TOKEN_2",
        json!({"name": "m", "code_up": "-- role analytics\nSELECT 1"}),
    )
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    // An admin's migration re-pushed as stored writes nothing, so it passes. Replacing it removes
    // SQL they could not run, even with SQL they could, so it is refused like deleting it.
    let resp = upsert("SECRET_TOKEN", "DROP TABLE t").await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    let resp = upsert("SECRET_TOKEN_2", "DROP TABLE t").await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    for code_up in ["DROP TABLE u", "-- role analytics\nSELECT 1"] {
        let resp = upsert("SECRET_TOKEN_2", code_up).await?;
        assert_eq!(resp.status(), 401, "{code_up}: {}", resp.text().await?);
    }
    let resp = authed(
        client().delete(format!(
            "{base}/delete_datatable_migration/main/20260101000000"
        )),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 401, "{}", resp.text().await?);

    let code_up: String = sqlx::query_scalar(
        "SELECT code_up FROM datatable_migrations
         WHERE workspace_id = 'test-workspace' AND timestamp = 20260101000000",
    )
    .fetch_one(&db)
    .await?;
    assert_eq!(code_up, "DROP TABLE t");
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn an_operator_neither_writes_nor_runs_a_migration(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    enable_migrations_on_main(&db).await?;
    // Off roles, where every role check passes everyone, so only the operator guard is left to
    // refuse.
    sqlx::query(
        "UPDATE workspace_settings SET datatable = datatable #- '{datatables,main,permissions}'
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;
    // A token of its own: the auth cache is process-wide, and the other tests here authenticate
    // `SECRET_TOKEN_3` in this workspace as a non-operator.
    sqlx::raw_sql(
        "UPDATE usr SET operator = true, role = 'Operator'
         WHERE workspace_id = 'test-workspace' AND username = 'test-user-3';
         INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin)
         VALUES (encode(sha256('DT_OPERATOR_TOKEN'::bytea), 'hex'), 'DT_OPERATO',
                 'DT_OPERATOR_TOKEN', 'test3@windmill.dev', 'operator', false);",
    )
    .execute(&db)
    .await?;
    let server = ApiServer::start(db.clone()).await?;
    let base = format!(
        "http://localhost:{}/api/w/test-workspace/workspaces",
        server.addr.port()
    );
    let migration = json!({"timestamp": 20260101000000i64, "name": "m", "code_up": "SELECT 1"});

    for request in [
        client()
            .post(format!("{base}/create_datatable_migration/main"))
            .json(&migration),
        client()
            .post(format!("{base}/upsert_datatable_migration/main"))
            .json(&migration),
        client().delete(format!(
            "{base}/delete_datatable_migration/main/20260101000000"
        )),
        client().post(format!("{base}/run_datatable_migrations/main")),
        client().post(format!("{base}/rollback_datatable_migrations/main")),
        client().post(format!("{base}/generate_initial_datatable_migration/main")),
    ] {
        let resp = authed(request, "DT_OPERATOR_TOKEN").send().await?;
        let status = resp.status();
        let text = resp.text().await?;
        assert!(
            status == 401 && text.contains("Operators cannot"),
            "{status}: {text}"
        );
    }

    // The same write from a developer is accepted: what refused the operator is theirs alone.
    let resp = authed(
        client()
            .post(format!("{base}/upsert_datatable_migration/main"))
            .json(&migration),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    Ok(())
}

#[cfg(not(all(feature = "private", feature = "enterprise")))]
const ENTERPRISE_REFUSAL: &str = "Data table roles are a Windmill Enterprise Edition feature";

#[cfg(not(all(feature = "private", feature = "enterprise")))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn every_roles_route_is_an_enterprise_feature(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let api = format!("http://localhost:{}/api", server.addr.port());
    let dt = format!("{api}/w/test-workspace/workspaces");

    for (method, url, body) in [
        (
            reqwest::Method::GET,
            format!("{api}/settings/datatable_roles"),
            None,
        ),
        (
            reqwest::Method::POST,
            format!("{api}/settings/datatable_roles"),
            Some(json!({ "name": "wmtest_ce" })),
        ),
        (
            reqwest::Method::POST,
            format!("{api}/settings/datatable_roles/role1"),
            Some(json!({ "enabled": false })),
        ),
        (
            reqwest::Method::DELETE,
            format!("{api}/settings/datatable_roles/role1"),
            None,
        ),
        (
            reqwest::Method::GET,
            format!("{dt}/datatable_permissions/main"),
            None,
        ),
        (
            reqwest::Method::POST,
            format!("{dt}/datatable_permissions/main"),
            Some(json!({ "permissioned": false })),
        ),
        (
            reqwest::Method::GET,
            format!("{dt}/datatable_usable_roles/main"),
            None,
        ),
    ] {
        let mut request = authed(client().request(method.clone(), &url), "SECRET_TOKEN");
        if let Some(body) = body {
            request = request.json(&body);
        }
        let resp = request.send().await?;
        let status = resp.status();
        let text = resp.text().await?;
        assert!(
            status == 400 && text.contains(ENTERPRISE_REFUSAL),
            "{method} {url} answered {status}: {text}"
        );
    }

    // Refused, not acted on: the catalog row and the data table's roles are where they were.
    let untouched: (i64, bool) = sqlx::query_as(
        "SELECT (SELECT count(*) FROM datatable_role),
                (datatable->'datatables'->'main') ? 'permissions'
         FROM workspace_settings WHERE workspace_id = 'test-workspace'",
    )
    .fetch_one(&db)
    .await?;
    assert_eq!(untouched, (1, true));
    Ok(())
}

#[cfg(not(all(feature = "private", feature = "enterprise")))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn without_the_enterprise_edition_a_data_table_under_roles_is_refused_a_connection(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    use windmill_common::workspaces::{get_datatable_resource_from_db, DatatableAccess};
    initialize_tracing().await;

    // Saved under roles, as an enterprise build left it: refused whoever asks, never `admin`.
    for access in [DatatableAccess::Unchecked, DatatableAccess::NoIdentity] {
        let err = get_datatable_resource_from_db(&db, "test-workspace", "main", None, access)
            .await
            .expect_err("a data table under roles resolved");
        assert!(err.to_string().contains(ENTERPRISE_REFUSAL), "{err}");
    }

    // Not under roles, it resolves as it always has, including when `admin` is named — which every
    // migration does; naming any other role on it is refused.
    sqlx::query(
        "UPDATE workspace_settings SET datatable = datatable #- '{datatables,main,permissions}'
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;
    for role in [None, Some("admin")] {
        let resolved = get_datatable_resource_from_db(
            &db,
            "test-workspace",
            "main",
            role,
            DatatableAccess::NoIdentity,
        )
        .await?;
        assert_eq!(resolved["dbname"], "dt_main", "{resolved}");
    }
    let err = get_datatable_resource_from_db(
        &db,
        "test-workspace",
        "main",
        Some("analytics"),
        DatatableAccess::Unchecked,
    )
    .await
    .expect_err("a named role resolved");
    assert!(err.to_string().contains(ENTERPRISE_REFUSAL), "{err}");
    Ok(())
}

/// A save blocked on an instance database's lock is not a user of it until it commits one: the
/// cleanup for a database whose setup failed lets that save through and reads what it left.
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn cleanup_waits_out_a_save_racing_it_for_an_instance_database(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let key = "instance_database:dt_probe";

    // Queues cleanup behind `holder`, then a save behind cleanup, so cleanup takes the lock with
    // the save already waiting on it — the ordering the waiter check is for.
    let race = |commit: bool| {
        let db = db.clone();
        async move {
            let mut holder = db.acquire().await?;
            sqlx::query("SELECT pg_advisory_lock(hashtext($1))")
                .bind(key)
                .execute(&mut *holder)
                .await?;

            let cleanup = tokio::spawn({
                let db = db.clone();
                async move { windmill_common::drop_unused_instance_database(&db, "dt_probe").await }
            });
            tokio::time::sleep(std::time::Duration::from_millis(300)).await;

            let save = tokio::spawn({
                let db = db.clone();
                async move {
                    let mut tx = db.begin().await?;
                    windmill_common::lock_instance_databases(&mut tx, ["dt_probe"]).await?;
                    sqlx::query(
                        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable,
                             '{datatables,probe}',
                             '{"database": {"resource_type": "instance", "resource_path": "dt_probe"}}')
                           WHERE workspace_id = 'test-workspace'"#,
                    )
                    .execute(&mut *tx)
                    .await?;
                    if commit {
                        tx.commit().await?;
                    } else {
                        tx.rollback().await?;
                    }
                    Ok::<_, anyhow::Error>(())
                }
            });
            tokio::time::sleep(std::time::Duration::from_millis(300)).await;

            sqlx::query("SELECT pg_advisory_unlock(hashtext($1))")
                .bind(key)
                .execute(&mut *holder)
                .await?;
            save.await??;
            Ok::<_, anyhow::Error>(cleanup.await??)
        }
    };

    assert!(
        matches!(race(false).await?, windmill_common::Cleanup::Dropped),
        "a save that rolled back kept the database, whose name then blocks every retry"
    );
    let kept = race(true).await?;
    let windmill_common::Cleanup::InUse(users) = kept else {
        anyhow::bail!("the database was dropped under a save that committed a reference to it")
    };
    assert_eq!(users, vec!["test-workspace".to_string()]);
    Ok(())
}

#[cfg(all(feature = "private", feature = "enterprise"))]
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_connection_naming_no_role_holds_the_default_one_and_the_viewers(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    use windmill_common::workspaces::{get_datatable_resource_from_db, DatatableAccess};

    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let suffix: String = uuid::Uuid::new_v4().simple().to_string()[..8].to_string();
    let names = [
        format!("wmtest_floor_{suffix}"),
        format!("wmtest_emea_{suffix}"),
        format!("wmtest_amer_{suffix}"),
    ];
    let roles_url = format!("http://localhost:{port}/api/settings/datatable_roles");
    const MEMBER_LOGINS: &str = "SELECT m.rolname::text FROM pg_auth_members am
         JOIN pg_roles r ON r.oid = am.roleid JOIN pg_roles m ON m.oid = am.member
         WHERE r.rolname = $1 AND m.rolname LIKE 'wm\\_set\\_%'";

    // The job is the app publisher's in every case: only the viewer the server recorded differs.
    let connect = |role: Option<&'static str>, viewer: Option<&'static str>| {
        get_datatable_resource_from_db(
            &db,
            "test-workspace",
            "main",
            role,
            DatatableAccess::PermissionedAs {
                permissioned_as: "u/test-user",
                email: "test@windmill.dev",
                end_user_email: viewer,
            },
        )
    };
    let set_permissions = |default_role: String, roles: Value| {
        sqlx::query(
            "UPDATE workspace_settings
             SET datatable = jsonb_set(datatable, '{datatables,main,permissions}', $1)
             WHERE workspace_id = 'test-workspace'",
        )
        .bind(json!({ "default_role": default_role, "roles": roles }))
        .execute(&db)
    };

    let outcome = async {
        let mut ids = vec![];
        for name in &names {
            let created: Value = authed(client().post(&roles_url), "SECRET_TOKEN")
                .json(&json!({ "name": name }))
                .send()
                .await?
                .error_for_status()?
                .json()
                .await?;
            ids.push(created["id"].as_str().unwrap().to_string());
        }
        // `test-user-2` is named by both regional roles, one of them through a group, and by
        // `admin`; `test-user-3` by one; the workspace admin `test-user` by neither.
        let roles = json!({
            "admin": {"tenants": ["u/test-user-2"]},
            (ids[0].as_str()): {"tenants": ["*"]},
            (ids[1].as_str()): {"tenants": ["u/test-user-2", "u/test-user-3"]},
            (ids[2].as_str()): {"tenants": ["g/analysts"]},
        });
        set_permissions(ids[0].clone(), roles.clone()).await?;
        sqlx::query(
            "INSERT INTO usr_to_group (workspace_id, usr, group_)
             VALUES ('test-workspace', 'test-user-2', 'analysts')",
        )
        .execute(&db)
        .await?;
        let member_of = |login: String| async {
            let mut roles: Vec<String> = sqlx::query_scalar(
                "SELECT r.rolname::text FROM pg_auth_members am
                 JOIN pg_roles r ON r.oid = am.roleid JOIN pg_roles m ON m.oid = am.member
                 WHERE m.rolname = $1",
            )
            .bind(login)
            .fetch_all(&db)
            .await?;
            roles.sort();
            Ok::<_, anyhow::Error>(roles)
        };

        let all = connect(None, Some("test2@windmill.dev")).await?;
        let login = all["user"].as_str().unwrap().to_string();
        assert!(login.starts_with("wm_set_"), "{login}");
        let mut expected = names.to_vec();
        expected.sort();
        assert_eq!(
            member_of(login.clone()).await?,
            expected,
            "the login holds the default role and the viewer's, and not `admin`"
        );
        assert_eq!(connect(None, Some("test2@windmill.dev")).await?, all);

        let two = connect(None, Some("test3@windmill.dev")).await?;
        let mut expected = names[..2].to_vec();
        expected.sort();
        assert_eq!(
            member_of(two["user"].as_str().unwrap().to_string()).await?,
            expected
        );

        // An admin may use every role, the one owning the tables included. As a viewer they add
        // what names them, which here is nothing; nor does a job with no viewer recorded.
        for viewer in [Some("test@windmill.dev"), None] {
            assert_eq!(connect(None, viewer).await?["user"], names[0].as_str());
        }
        // A named role is that role alone, whoever is viewing.
        let named: &'static str = Box::leak(names[1].clone().into_boxed_str());
        assert_eq!(
            connect(Some(named), Some("test2@windmill.dev")).await?["user"],
            names[1].as_str()
        );
        // A default of `admin` is the connection it always was: the viewer's roles are not added.
        set_permissions("admin".to_string(), roles.clone()).await?;
        let admin = connect(None, Some("test2@windmill.dev")).await?;
        assert_eq!(admin, connect(Some("admin"), None).await?);
        assert!(!admin["user"].as_str().unwrap().starts_with("wm_"));

        // With no default role the viewer's roles are all there is, the one open to everyone
        // included, and a connection left with none is refused rather than resolved as anything.
        set_permissions("$none".to_string(), roles.clone()).await?;
        assert_eq!(
            connect(None, Some("test3@windmill.dev")).await?,
            two,
            "no default changes nothing for a viewer the roles name"
        );
        assert!(connect(None, None).await.is_err());
        // The same holds for a default the job's own identity is not a tenant of.
        let mut closed = roles.clone();
        closed[ids[0].as_str()] = json!({"tenants": []});
        set_permissions(ids[0].clone(), closed).await?;
        let as_member = |viewer: Option<&'static str>| {
            get_datatable_resource_from_db(
                &db,
                "test-workspace",
                "main",
                None,
                DatatableAccess::PermissionedAs {
                    permissioned_as: "u/test-user-3",
                    email: "test3@windmill.dev",
                    end_user_email: viewer,
                },
            )
        };
        assert_eq!(
            as_member(Some("test3@windmill.dev")).await?["user"],
            names[1].as_str()
        );
        assert!(as_member(None).await.is_err());
        set_permissions(ids[0].clone(), roles).await?;

        // A set login nobody connects with any more is dropped, a recent one is kept, and a
        // dropped one is made again for the next viewer who needs it.
        let exists = || async {
            sqlx::query_scalar::<_, bool>(
                "SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)",
            )
            .bind(&login)
            .fetch_one(&db)
            .await
        };
        windmill_common::datatable_roles::sweep_unused_role_set_logins(&db).await?;
        assert!(exists().await?, "a set login in use was swept");
        sqlx::query("UPDATE datatable_role_set SET last_used_at = now() - interval '31 days'")
            .execute(&db)
            .await?;
        windmill_common::datatable_roles::sweep_unused_role_set_logins(&db).await?;
        assert!(!exists().await?, "an unused set login was kept");
        assert_eq!(
            connect(None, Some("test2@windmill.dev")).await?["user"],
            login.as_str()
        );

        // A disabled role must not stay reachable through a login that is a member of it.
        authed(
            client().post(format!("{roles_url}/{}", ids[2])),
            "SECRET_TOKEN",
        )
        .json(&json!({ "enabled": false }))
        .send()
        .await?
        .error_for_status()?;
        let can_login: bool =
            sqlx::query_scalar("SELECT rolcanlogin FROM pg_roles WHERE rolname = $1")
                .bind(&login)
                .fetch_one(&db)
                .await?;
        assert!(!can_login, "{login} still logs in with a disabled role");
        assert_eq!(
            connect(None, Some("test2@windmill.dev")).await?["user"],
            two["user"]
        );
        Ok::<_, anyhow::Error>(())
    }
    .await;

    // Roles are cluster-wide, so they outlive this test's throwaway database.
    for name in &names {
        let logins: Vec<String> = sqlx::query_scalar(MEMBER_LOGINS)
            .bind(name)
            .fetch_all(&db)
            .await
            .unwrap_or_default();
        for login in logins {
            let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{login}\""))
                .execute(&db)
                .await;
        }
        let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{name}\""))
            .execute(&db)
            .await;
    }
    outcome
}