awatertrevi/infisical
1
0
Fork 0
mirror of https://github.com/awatertrevi/infisical.git synced 2026-09-22 13:39:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use mock api call

Browse Source
This commit is contained in:
Fang-Pen Lin
2025-11-12 21:26:43 -08:00
parent 146524ac2d
commit 1b90a1dca0
4 changed files with 154 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend/bdd/features/environment.py
View File

@@ -10,6 +10,8 @@ from dotenv import load_dotenv
from faker import Faker
import logging
from features.steps.utils import clear_all_nock, restore_nock
load_dotenv()
logger = logging.getLogger(__name__)
@@ -205,3 +207,5 @@ def before_all(context: Context):
def after_scenario(context: Context, scenario: typing.Any):
if hasattr(context, "web_server"):
context.web_server.shutdown_and_server_close()
clear_all_nock(context)
restore_nock(context)

103
backend/bdd/features/steps/pki_acme.py
View File

@@ -22,7 +22,7 @@ from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes
from utils import replace_vars
from utils import replace_vars, with_nocks
from utils import eval_var
from utils import prepare_headers
@@ -87,6 +87,107 @@ def step_impl(context: Context, profile_var: str):
)
@given("I create a Cloudflare connection as {var_name}")
def step_impl(context: Context, var_name: str):
jwt_token = context.vars["AUTH_TOKEN"]
conn_slug = faker.slug()
mock_account_id = "MOCK_ACCOUNT_ID"
with with_nocks(
context,
definitions=[
{
"scope": "https://api.cloudflare.com:443",
"method": "GET",
"path": f"/client/v4/accounts/{mock_account_id}",
"status": 200,
"response": {
"result": {
"id": "A2A6347F-88B5-442D-9798-95E408BC7701",
"name": "Mock Account",
"type": "standard",
"settings": {
"enforce_twofactor": True,
"api_access_enabled": None,
"access_approval_expiry": None,
"abuse_contact_email": None,
"user_groups_ui_beta": False,
},
"legacy_flags": {
"enterprise_zone_quota": {
"maximum": 0,
"current": 0,
"available": 0,
}
},
"created_on": "2013-04-18T00:41:02.215243Z",
},
"success": True,
"errors": [],
"messages": [],
},
"responseIsBinary": False,
}
],
):
response = context.http_client.post(
"/api/v1/app-connections/cloudflare",
headers=dict(authorization="Bearer {}".format(jwt_token)),
json={
"name": conn_slug,
"description": "",
"method": "api-token",
"credentials": {
"apiToken": "MOCK_API_TOKEN",
"accountId": mock_account_id,
},
},
)
response.raise_for_status()
context.vars[var_name] = response
@given('I have an ACME cert profile with external ACME CA as "{profile_var}"')
def step_impl(context: Context, profile_var: str):
profile_id = context.vars.get("PROFILE_ID")
secret = context.vars.get("EAB_SECRET")
if profile_id is not None and secret is not None:
kid = profile_id
else:
profile_slug = faker.slug()
jwt_token = context.vars["AUTH_TOKEN"]
response = context.http_client.post(
"/api/v1/pki/certificate-profiles",
headers=dict(authorization="Bearer {}".format(jwt_token)),
json={
"projectId": context.vars["PROJECT_ID"],
"slug": profile_slug,
"description": "ACME Profile created by BDD test",
"enrollmentType": "acme",
"caId": context.vars["CERT_CA_ID"],
"certificateTemplateId": context.vars["CERT_TEMPLATE_ID"],
"acmeConfig": {},
},
)
response.raise_for_status()
resp_json = response.json()
profile_id = resp_json["certificateProfile"]["id"]
kid = profile_id
response = context.http_client.get(
f"/api/v1/pki/certificate-profiles/{profile_id}/acme/eab-secret/reveal",
headers=dict(authorization="Bearer {}".format(jwt_token)),
)
response.raise_for_status()
resp_json = response.json()
secret = resp_json["eabSecret"]
context.vars[profile_var] = AcmeProfile(
profile_id,
eab_kid=kid,
eab_secret=secret,
)
@given("I use {token_var} for authentication")
def step_impl(context: Context, token_var: str):
context.auth_token = eval_var(context, token_var)

41
backend/bdd/features/steps/utils.py
View File

@@ -3,6 +3,7 @@ from cryptography.hazmat.primitives import hashes
from cryptography.x509.oid import NameOID
import logging
import re
import contextlib
import httpx
import requests
@@ -258,3 +259,43 @@ def x509_cert_to_dict(cert: x509.Certificate) -> dict:
)
return result
def define_nock(context: Context, definitions: list[dict]):
jwt_token = context.vars["AUTH_TOKEN"]
response = context.http_client.post(
"/api/v1/bdd-nock/define",
headers=dict(authorization="Bearer {}".format(jwt_token)),
json=dict(definitions=definitions),
)
response.raise_for_status()
def restore_nock(context: Context):
jwt_token = context.vars["AUTH_TOKEN"]
response = context.http_client.post(
"/api/v1/bdd-nock/restore",
headers=dict(authorization="Bearer {}".format(jwt_token)),
json=dict(),
)
response.raise_for_status()
def clear_all_nock(context: Context):
jwt_token = context.vars["AUTH_TOKEN"]
response = context.http_client.post(
"/api/v1/bdd-nock/clear-all",
headers=dict(authorization="Bearer {}".format(jwt_token)),
json=dict(),
)
response.raise_for_status()
@contextlib.contextmanager
def with_nocks(context: Context, definitions: list[dict]):
try:
define_nock(context, definitions)
yield
finally:
clear_all_nock(context)
restore_nock(context)

10
backend/src/server/routes/v1/bdd-nock-router.ts
View File

@@ -4,6 +4,7 @@ import { getConfig } from "@app/lib/config/env";
import { ForbiddenRequestError } from "@app/lib/errors";
import { verifyAuth } from "@app/server/plugins/auth/verify-auth";
import { AuthMode } from "@app/services/auth/auth-type";
import { logger } from "@app/lib/logger";
import nock, { Definition } from "nock";
export const registerBddNockRouter = async (server: FastifyZodProvider) => {
@@ -20,7 +21,7 @@ export const registerBddNockRouter = async (server: FastifyZodProvider) => {
method: "POST",
url: "/define",
schema: {
body: z.object({ definition: z.string() }),
body: z.object({ definitions: z.unknown().array() }),
response: {
200: z.object({ status: z.string() })
}
@@ -29,8 +30,9 @@ export const registerBddNockRouter = async (server: FastifyZodProvider) => {
handler: async (req) => {
checkIfBddNockApiEnabled();
const { body } = req;
const { definition } = body;
nock.define(definition as unknown as Definition[]);
const { definitions } = body;
logger.info(definitions, "Defining nock");
nock.define(definitions as Definition[]);
return { status: "ok" };
}
});
@@ -46,6 +48,7 @@ export const registerBddNockRouter = async (server: FastifyZodProvider) => {
onRequest: verifyAuth([AuthMode.JWT]),
handler: async (req) => {
checkIfBddNockApiEnabled();
logger.info("Restore network requests from nock");
nock.restore();
return { status: "ok" };
}
@@ -62,6 +65,7 @@ export const registerBddNockRouter = async (server: FastifyZodProvider) => {
onRequest: verifyAuth([AuthMode.JWT]),
handler: async (req) => {
checkIfBddNockApiEnabled();
logger.info("Cleaning all nocks");
nock.cleanAll();
return { status: "ok" };
}