Skip to content

API reference

adc_locust.env_config

Load, validate, and interactively create the .env file that supplies Nitro API credentials and NetScaler address information.

Credentials are never hardcoded or logged. Values are read from a gitignored .env file in the current working directory (or a path given via --env-file / ADC_LOCUST_ENV_FILE), following the same pattern used by the reference Nitro SDK scripts this project is built from.

EnvConfigError

Bases: RuntimeError

Raised when required Nitro connection settings are missing.

NitroEnvConfig(nitro_user, nitro_pass, nitro_url, verify_tls=False) dataclass

Resolved Nitro API connection settings.

default_env_path()

Return the default .env location: ADC_LOCUST_ENV_FILE or CWD/.env.

Source code in src/adc_locust/env_config.py
55
56
57
58
59
def default_env_path() -> Path:
    """Return the default `.env` location: `ADC_LOCUST_ENV_FILE` or CWD/.env."""
    if "ADC_LOCUST_ENV_FILE" in os.environ:
        return Path(os.environ["ADC_LOCUST_ENV_FILE"])
    return Path.cwd() / ".env"

load_env(path=None)

Load and validate Nitro connection settings from a .env file.

Raises EnvConfigError with a human-readable explanation (including the contents to add) if the file is missing or incomplete.

Source code in src/adc_locust/env_config.py
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def load_env(path: Path | None = None) -> NitroEnvConfig:
    """Load and validate Nitro connection settings from a `.env` file.

    Raises EnvConfigError with a human-readable explanation (including the
    contents to add) if the file is missing or incomplete.
    """
    env_path = path or default_env_path()
    missing = missing_variables(env_path)
    if missing:
        reason = "does not exist" if not env_path.exists() else f"is missing {', '.join(missing)}"
        raise EnvConfigError(f"{env_path} {reason}.\n\n{ENV_FILE_HELP.format(path=env_path)}")

    load_dotenv(env_path, override=True)
    values = dotenv_values(env_path)
    return NitroEnvConfig(
        nitro_user=values["NITRO_USER"],  # type: ignore[arg-type]
        nitro_pass=values["NITRO_PASS"],  # type: ignore[arg-type]
        nitro_url=values["NITRO_URL"],  # type: ignore[arg-type]
        verify_tls=_parse_bool(values.get("NITRO_VERIFY_TLS"), OPTIONAL_VARS["NITRO_VERIFY_TLS"] == "true"),
    )

missing_variables(path)

Return the required variable names not present (or empty) in path.

Source code in src/adc_locust/env_config.py
68
69
70
71
72
73
def missing_variables(path: Path) -> list[str]:
    """Return the required variable names not present (or empty) in `path`."""
    if not path.exists():
        return list(REQUIRED_VARS)
    values = dotenv_values(path)
    return [name for name in REQUIRED_VARS if not values.get(name)]

write_env(path, *, nitro_user, nitro_pass, nitro_url, verify_tls=False)

Create or overwrite path with the given Nitro connection settings.

Used by the interactive setup screen so a .env can be produced without leaving the Textual interface. The file is written with owner-only permissions since it holds a plaintext credential.

Source code in src/adc_locust/env_config.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def write_env(
    path: Path,
    *,
    nitro_user: str,
    nitro_pass: str,
    nitro_url: str,
    verify_tls: bool = False,
) -> None:
    """Create or overwrite `path` with the given Nitro connection settings.

    Used by the interactive setup screen so a `.env` can be produced without
    leaving the Textual interface. The file is written with owner-only
    permissions since it holds a plaintext credential.
    """
    content = (
        "# Created by adc-locust's interactive setup screen.\n"
        "# Nitro API credentials and NetScaler address information.\n"
        f"NITRO_USER={nitro_user}\n"
        f"NITRO_PASS={nitro_pass}\n"
        f"NITRO_URL={nitro_url}\n"
        f"NITRO_VERIFY_TLS={'true' if verify_tls else 'false'}\n"
    )
    path.write_text(content, encoding="utf-8")
    with contextlib.suppress(OSError):
        path.chmod(0o600)  # Best-effort on platforms/filesystems that don't support chmod.

adc_locust.nitro_client

Read-only Nitro API client used to discover NetScaler load-balancing and content-switching configuration available for load testing.

Built from the patterns in list-vservers-multi-service.py, check-cs-targets.py, and dot_4485/locust/common/distribution_check.py (nitro reference project). All operations here are read-only queries against the ADC — no configuration is created, modified, or deleted.

BoundService(name, weight=None) dataclass

One service bound to an LB vserver.

CSPolicyBinding(policy_name, priority, rule, target_lbvserver) dataclass

One Content Switching policy bound to a CS vserver.

CSVServerInfo(name, ip, port, servicetype, state, default_lbvserver, policy_bindings=list()) dataclass

A Content Switching vserver, its default target, and its policies.

target_for(lb_vserver_name)

Return the policy binding routing to lb_vserver_name, if any.

Source code in src/adc_locust/nitro_client.py
101
102
103
104
105
106
def target_for(self, lb_vserver_name: str) -> CSPolicyBinding | None:
    """Return the policy binding routing to `lb_vserver_name`, if any."""
    for binding in self.policy_bindings:
        if binding.target_lbvserver == lb_vserver_name:
            return binding
    return None

NitroClient(env)

A read-only session against a NetScaler's Nitro API.

Use as a context manager so the session is always logged out:

with NitroClient(env) as client:
    vservers = client.list_lb_vservers()
Source code in src/adc_locust/nitro_client.py
121
122
123
124
125
def __init__(self, env: NitroEnvConfig) -> None:
    self._env = env
    self._session: nitro_service | None = None
    if not env.verify_tls:
        urllib3.disable_warnings(category=urllib3.exceptions.InsecureRequestWarning)

connect()

Log in to the Nitro API. Raises NitroConnectionError on failure.

Source code in src/adc_locust/nitro_client.py
127
128
129
130
131
132
133
134
135
136
137
def connect(self) -> None:
    """Log in to the Nitro API. Raises NitroConnectionError on failure."""
    session = nitro_service(connect_context=self._env.nitro_url, protocol="https")
    session.certvalidation = self._env.verify_tls
    try:
        session.login(username=self._env.nitro_user, password=self._env.nitro_pass, timeout=3600)
    except Exception as error:  # nitro_service raises its own exception types
        raise NitroConnectionError(f"Could not log in to {self._env.nitro_url}: {error}") from error
    if not session.isLogin():
        raise NitroConnectionError(f"Login to {self._env.nitro_url} was rejected.")
    self._session = session

find_cs_target(lb_vserver_name)

Find the CS vserver (and matching policy, if any) that routes to an unroutable LB vserver. Returns None if no CS vserver targets it.

Source code in src/adc_locust/nitro_client.py
233
234
235
236
237
238
239
240
241
242
def find_cs_target(self, lb_vserver_name: str) -> tuple[CSVServerInfo, CSPolicyBinding | None] | None:
    """Find the CS vserver (and matching policy, if any) that routes to
    an unroutable LB vserver. Returns None if no CS vserver targets it."""
    for cs in self.list_cs_vservers():
        if cs.default_lbvserver == lb_vserver_name:
            return cs, None
        target = cs.target_for(lb_vserver_name)
        if target is not None:
            return cs, target
    return None

list_cs_vservers()

Return all Content Switching vservers with their policy bindings.

Source code in src/adc_locust/nitro_client.py
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def list_cs_vservers(self) -> list[CSVServerInfo]:
    """Return all Content Switching vservers with their policy bindings."""
    session = self._require_session
    results: list[CSVServerInfo] = []
    for cs in csvserver.get(session) or []:
        cs_name = _get(cs, "name")
        try:
            bindings = csvserver_cspolicy_binding.get(session, cs_name)
        except Exception:
            bindings = None

        policy_bindings: list[CSPolicyBinding] = []
        if bindings:
            pol_bindings = bindings if isinstance(bindings, list) else [bindings]
            for pb in pol_bindings:
                policy_bindings.append(
                    CSPolicyBinding(
                        policy_name=_get(pb, "policyname"),  # type: ignore[arg-type]
                        priority=_get(pb, "priority"),  # type: ignore[arg-type]
                        rule=_get(pb, "rule"),  # type: ignore[arg-type]
                        target_lbvserver=_get(pb, "targetlbvserver"),  # type: ignore[arg-type]
                    )
                )

        results.append(
            CSVServerInfo(
                name=cs_name,  # type: ignore[arg-type]
                ip=_get(cs, "ipv46", "") or "",  # type: ignore[arg-type]
                port=str(_get(cs, "port", "")),
                servicetype=_get(cs, "servicetype", ""),  # type: ignore[arg-type]
                state=_get(cs, "curstate", "N/A"),  # type: ignore[arg-type]
                default_lbvserver=_get(cs, "lbvserver"),  # type: ignore[arg-type]
                policy_bindings=policy_bindings,
            )
        )
    return results

list_lb_vservers(*, multi_service_only=True)

Return LB vservers, optionally limited to those balancing traffic across more than one bound service (real load-balancing scenarios).

Source code in src/adc_locust/nitro_client.py
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def list_lb_vservers(self, *, multi_service_only: bool = True) -> list[VServerInfo]:
    """Return LB vservers, optionally limited to those balancing traffic
    across more than one bound service (real load-balancing scenarios)."""
    session = self._require_session
    results: list[VServerInfo] = []
    for vs in lbvserver.get(session) or []:
        services = self._bound_services(vs.name)
        info = VServerInfo(
            name=vs.name,
            ip=getattr(vs, "ipv46", "") or "",
            port=str(getattr(vs, "port", "")),
            servicetype=getattr(vs, "servicetype", ""),
            state=f"{getattr(vs, 'curstate', 'N/A')}/{getattr(vs, 'status', 'N/A')}",
            lb_method=getattr(vs, "lbmethod", "N/A"),
            persistence_type=getattr(vs, "persistencetype", "NONE"),
            persistence_timeout=getattr(vs, "timeout", None),
            bound_services=services,
        )
        if not multi_service_only or info.has_multiple_services:
            results.append(info)
    return results

snapshot_service_hits(service_names)

Return {service_name: totalrequests} for the given bound services.

Used to compare traffic distribution before/after a Locust run. Read-only — queries service_stats for each named service.

Source code in src/adc_locust/nitro_client.py
244
245
246
247
248
249
250
251
252
253
254
255
256
def snapshot_service_hits(self, service_names: list[str]) -> dict[str, int]:
    """Return {service_name: totalrequests} for the given bound services.

    Used to compare traffic distribution before/after a Locust run.
    Read-only — queries service_stats for each named service.
    """
    session = self._require_session
    snapshot: dict[str, int] = {}
    for name in service_names:
        stats = service_stats.get(session, name)
        hits = _get(stats, "totalrequests", 0)
        snapshot[name] = int(hits) if hits is not None else 0  # type: ignore[call-overload]
    return snapshot

NitroConnectionError

Bases: RuntimeError

Raised when logging in to the NetScaler Nitro API fails.

VServerInfo(name, ip, port, servicetype, state, lb_method, persistence_type, persistence_timeout, bound_services=list()) dataclass

A load-balancing virtual server and the services it balances across.

is_routable property

Whether this vserver has its own IP, or must be reached via CS.

adc_locust.locust_runner

Build and run Locust load tests against a discovered NetScaler vserver.

Targets are derived from Nitro API discovery (see nitro_client.py) rather than hardcoded, unlike the reference dot_4485/locust/common/config.py which lists specific known vservers. Locust itself is invoked headless as a subprocess, matching the documented usage in the reference project (locust -f locustfile.py --tags fairness), so Locust's gevent runtime never shares a process/event loop with the Textual UI.

LoadTestConfig(users=10, spawn_rate=2.0, run_time_seconds=30, mode=FAIRNESS) dataclass

Locust run parameters for a single test.

LoadTestError

Bases: RuntimeError

Raised when Locust cannot be run or its results cannot be parsed.

LoadTestResult(request_count, failure_count, requests_per_second, median_response_time_ms, service_deltas, warnings) dataclass

Aggregated Locust stats plus the per-service distribution delta.

ServiceDelta(service_name, before, after) dataclass

Hit-count change for one bound service across a test run.

TestTarget(lb_vserver_name, target_host, target_port, scheme, host_header, path_prefix, via_content_switching, persistence, bound_services=list()) dataclass

How to reach one vserver for load testing, and what it should show.

build_target(vserver, cs_match, *, port=443, scheme='https')

Derive a reachable test target for a discovered LB vserver.

If the vserver has its own routable IP, target it directly. Otherwise it must be a Content Switching target (see nitro_client.find_cs_target); route through the CS vserver's IP with the Host header / rule path needed to match its policy, mirroring dot_4485/locust/common/config.py.

Source code in src/adc_locust/locust_runner.py
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def build_target(
    vserver: VServerInfo,
    cs_match: tuple[CSVServerInfo, CSPolicyBinding | None] | None,
    *,
    port: int = 443,
    scheme: str = "https",
) -> TestTarget:
    """Derive a reachable test target for a discovered LB vserver.

    If the vserver has its own routable IP, target it directly. Otherwise
    it must be a Content Switching target (see nitro_client.find_cs_target);
    route through the CS vserver's IP with the Host header / rule path
    needed to match its policy, mirroring dot_4485/locust/common/config.py.
    """
    if vserver.is_routable:
        return TestTarget(
            lb_vserver_name=vserver.name,
            target_host=vserver.ip,
            target_port=int(vserver.port) if str(vserver.port).isdigit() else port,
            scheme=scheme,
            host_header=None,
            path_prefix="/",
            via_content_switching=False,
            persistence=vserver.persistence_type,
            bound_services=[s.name for s in vserver.bound_services],
        )

    if cs_match is None:
        raise LoadTestError(
            f"{vserver.name} has no routable IP and is not a target of any Content "
            "Switching vserver. It cannot be reached for load testing."
        )
    cs_vserver, policy = cs_match
    return TestTarget(
        lb_vserver_name=vserver.name,
        target_host=cs_vserver.ip,
        target_port=int(cs_vserver.port) if str(cs_vserver.port).isdigit() else port,
        scheme=scheme,
        host_header=policy.rule if policy and policy.rule else None,
        path_prefix="/",
        via_content_switching=True,
        persistence=vserver.persistence_type,
        bound_services=[s.name for s in vserver.bound_services],
    )

compare_service_hits(before, after)

Compute per-service deltas and flag services that received far less than their expected even share of new traffic (see dot_4485/locust/common/distribution_check.py).

Source code in src/adc_locust/locust_runner.py
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def compare_service_hits(
    before: dict[str, int], after: dict[str, int]
) -> tuple[list[ServiceDelta], list[str]]:
    """Compute per-service deltas and flag services that received far less
    than their expected even share of new traffic (see
    dot_4485/locust/common/distribution_check.py)."""
    deltas = [ServiceDelta(name, before.get(name, 0), after.get(name, 0)) for name in before]
    total = sum(max(d.delta, 0) for d in deltas)
    warnings: list[str] = []
    if total == 0:
        warnings.append("No new hits recorded — check that requests actually reached this vserver.")
        return deltas, warnings

    expected_share = 100.0 / len(deltas) if deltas else 0.0
    for d in deltas:
        pct = (d.delta / total * 100) if d.delta > 0 else 0.0
        if pct < expected_share * 0.5:
            warnings.append(
                f"{d.service_name} received far less than its expected "
                f"~{expected_share:.0f}% share ({pct:.1f}%)"
            )
    return deltas, warnings

generate_locustfile(target)

Render a standalone Locust scenario module for target.

Includes both a fairness User (clears cookies every request, to reveal the raw LB algorithm distribution) and a persistence User (retains cookies, to confirm ADC stickiness), tagged as in the reference scenarios so either can be selected with --tags.

Source code in src/adc_locust/locust_runner.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def generate_locustfile(target: TestTarget) -> str:
    """Render a standalone Locust scenario module for `target`.

    Includes both a fairness User (clears cookies every request, to reveal
    the raw LB algorithm distribution) and a persistence User (retains
    cookies, to confirm ADC stickiness), tagged as in the reference
    scenarios so either can be selected with `--tags`.
    """
    headers = f'{{"Host": {target.host_header!r}}}' if target.host_header else "{}"
    return f'''"""Generated Locust scenario for {target.lb_vserver_name}. Do not edit by
hand -- adc-locust regenerates this file for every test run."""

from locust import HttpUser, task, between, tag

HEADERS = {headers}
PATH = {target.path_prefix!r}


class FairnessUser(HttpUser):
    """Clears cookies before every request to reveal the raw LB distribution."""

    host = {target.base_url!r}
    wait_time = between(1, 2)

    @tag("fairness")
    @task
    def load(self):
        self.client.cookies.clear()
        self.client.get(PATH, headers=HEADERS, name="{target.lb_vserver_name} (fairness)")


class PersistenceUser(HttpUser):
    """Retains cookies for the life of a simulated user to confirm stickiness."""

    host = {target.base_url!r}
    wait_time = between(1, 2)

    def on_start(self):
        self.client.get(PATH, headers=HEADERS, name="{target.lb_vserver_name} (persistence-init)")

    @tag("persistence")
    @task
    def load(self):
        self.client.get(PATH, headers=HEADERS, name="{target.lb_vserver_name} (persistence)")
'''

run_load_test(target, config)

Run Locust headless against target and return aggregated results.

This call blocks for the duration of the run and should be invoked from a worker thread, not the UI thread. It does not itself query Nitro service stats — callers that want a distribution delta should snapshot with NitroClient.snapshot_service_hits() immediately before and after calling this, then pass both to compare_service_hits().

Source code in src/adc_locust/locust_runner.py
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def run_load_test(target: TestTarget, config: LoadTestConfig) -> LoadTestResult:
    """Run Locust headless against `target` and return aggregated results.

    This call blocks for the duration of the run and should be invoked from
    a worker thread, not the UI thread. It does not itself query Nitro
    service stats — callers that want a distribution delta should snapshot
    with NitroClient.snapshot_service_hits() immediately before and after
    calling this, then pass both to compare_service_hits().
    """
    if config.mode not in MODES:
        raise LoadTestError(f"mode must be one of {MODES}, got {config.mode!r}")

    with tempfile.TemporaryDirectory(prefix="adc-locust-") as tmpdir:
        locustfile = Path(tmpdir) / "locustfile.py"
        locustfile.write_text(generate_locustfile(target), encoding="utf-8")
        csv_prefix = Path(tmpdir) / "results"

        command = [
            sys.executable,
            "-m",
            "locust",
            "-f",
            str(locustfile),
            "--headless",
            "--host",
            target.base_url,
            "--users",
            str(config.users),
            "--spawn-rate",
            str(config.spawn_rate),
            "--run-time",
            f"{config.run_time_seconds}s",
            "--tags",
            config.mode,
            "--csv",
            str(csv_prefix),
            "--only-summary",
            "--loglevel",
            "ERROR",
        ]
        completed = subprocess.run(
            command, capture_output=True, text=True, timeout=config.run_time_seconds + 60
        )
        if completed.returncode not in (0, 1):  # 1 = failures occurred, still a valid run
            raise LoadTestError(f"locust exited with {completed.returncode}: {completed.stderr.strip()}")

        request_count, failure_count, rps, median_ms = _parse_stats_csv(csv_prefix)

    return LoadTestResult(
        request_count=request_count,
        failure_count=failure_count,
        requests_per_second=rps,
        median_response_time_ms=median_ms,
        service_deltas=[],
        warnings=[],
    )

adc_locust.app

ADC Locust: a Textual TUI for load-testing Citrix ADC load balancing and content switching using the Nitro API for discovery and Locust for traffic generation.

AdcLocustApp(env_path)

Bases: App[None]

Discover NetScaler vservers via Nitro, then load-test them with Locust.

Source code in src/adc_locust/app.py
23
24
25
def __init__(self, env_path: Path) -> None:
    super().__init__()
    self.env_path = env_path

main()

Run the application.

Source code in src/adc_locust/app.py
57
58
59
60
def main() -> None:
    """Run the application."""
    args = parse_args()
    AdcLocustApp(args.env_file).run()