Northstar Fleet Audit.

"""Northstar Fleet Audit: YOU implement the pure, offline audit engine.

Do not use network clients, OS commands, persistent writes, or global state here.
The browser capstone and local tests share this function signature.
"""
import csv
import io
from datetime import datetime

def build_audit(csv_text, fetch_page, settings, previous, seen):
    """Return the exact documented success or blocked dictionary.

    Block: {'ok': False, 'reason': REASON, 'previous': previous}
    Success: {'ok': True, 'state': 'planned' | 'simulated',
              'report': {'site': SITE, 'total': N, 'offline_hosts': SORTED},
              'new_keys': SORTED, 'previous': previous}

    REASON: config, approval, csv, duplicate, timeout,
            http-NNN, page, schema, mismatch.

    fetch_page(n) is an injected fictional responder. No live API calls.
    """
    # TODO: Move your completed browser solution here, refactor into helpers,
    #       and run the test suite. Returning fake success is not an answer.
    raise NotImplementedError("Implement the Northstar audit engine")

"""Fictional response sequences only: no HTTP and no external dependencies."""
from copy import deepcopy
from datetime import datetime, timezone, timedelta

BASE_NOW = datetime(2026, 11, 10, 2, tzinfo=timezone.utc)
BASE_CSV = "host,site\nSW-01,HQ\nAP-02,HQ\n"
APPROVAL = {
    "id": "CHG-1042", "site": "HQ", "approved": True,
    "start": BASE_NOW - timedelta(hours=1),
    "end": BASE_NOW + timedelta(hours=1),
}
SCENARIOS = (
    "complete", "empty", "duplicate", "timeout", "unavailable",
    "unauthorized", "throttled", "bad-page", "bad-schema",
    "expired", "replay",
)

class FakeFetcher:
    """A callable, repeatable page responder. No network connection."""
    def __init__(self, pages):
        self.pages = {page: list(replies) for page, replies in pages.items()}
        self.calls = []
    def __call__(self, number):
        self.calls.append(number)
        if number not in self.pages or not self.pages[number]:
            raise ValueError("unexpected-page")
        reply = self.pages[number].pop(0)
        if isinstance(reply, Exception):
            raise reply
        return deepcopy(reply)

def scenario(name='complete'):
    if name not in SCENARIOS:
        raise ValueError("unrecognized-scenario")
    csv_text, now, approval, seen = BASE_CSV, BASE_NOW, deepcopy(APPROVAL), set()
    pages = {
        1: [{"status":200,
             "devices":[{"host":"SW-01","status":"online"}],
             "next":2}],
        2: [{"status":200,
             "devices":[{"host":"AP-02","status":"offline"}],
             "next":None}],
    }
    if name == "empty":
        csv_text = "host,site\n"
        pages = {1:[{"status":200,"devices":[],"next":None}]}
    elif name == "duplicate":
        csv_text += "sw-01,REMOTE\n"
    elif name == "timeout":
        pages[1].insert(0, TimeoutError("fictional-timeout"))
    elif name == "unavailable":
        pages = {1:[{"status":503},{"status":503}]}
    elif name == "unauthorized":
        pages = {1:[{"status":401}]}
    elif name == "throttled":
        pages = {1:[{"status":429}]}
    elif name == "bad-page":
        pages = {1:[{"status":200,"devices":[],"next":1}]}
    elif name == "bad-schema":
        pages = {1:[{"status":200,
                      "devices":[{"host":"SW-01","status":"broken"}],
                      "next":None}]}
    elif name == "expired":
        now = approval["end"]
    elif name == "replay":
        seen.add("CHG-1042:review:AP-02")
    return csv_text, FakeFetcher(pages), approval, now, seen

"""Command-line wrapper for a LOCAL fictional audit; safe by default."""
import argparse
import json
import os
from pathlib import Path
import sys
import tempfile
from audit_core import build_audit
from fixtures import SCENARIOS, scenario

ROOT = Path(__file__).resolve().parent
REPORT_DIR = ROOT / "reports"
MAX_CSV_BYTES = 65536

def load_sample_csv():
    source = ROOT / "samples" / "inventory.csv"
    if not source.is_file() or source.stat().st_size > MAX_CSV_BYTES:
        raise ValueError("invalid-sample-file")
    return source.read_text(encoding="utf-8-sig")

def write_local_report(report, replace=False):
    """Explicit local-only output. No file is created by the default CLI."""
    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    destination = REPORT_DIR / "audit.json"
    if destination.exists() and not replace:
        raise FileExistsError("report-exists-use-replace")
    temp_path = None
    try:
        with tempfile.NamedTemporaryFile(
                "w", encoding="utf-8", dir=REPORT_DIR,
                prefix=".audit-", suffix=".tmp", delete=False) as handle:
            temp_path = Path(handle.name)
            json.dump(report, handle, indent=2, sort_keys=True)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        if destination.exists() and not replace:
            raise FileExistsError("report-exists-use-replace")
        os.replace(temp_path, destination)
        temp_path = None
        return destination
    finally:
        if temp_path is not None:
            temp_path.unlink(missing_ok=True)

def main(argv=None):
    parser = argparse.ArgumentParser(description="Northstar offline fleet audit")
    parser.add_argument("--scenario", choices=SCENARIOS, default="complete")
    parser.add_argument("--mode", choices=("plan","simulate"), default="plan")
    parser.add_argument("--site", default="HQ")
    parser.add_argument("--max-pages", type=int, default=3)
    parser.add_argument("--save-report", action="store_true",
                        help="explicitly save LOCAL verified JSON under reports/")
    parser.add_argument("--replace", action="store_true",
                        help="allow replacing an existing LOCAL report with --save-report")
    args = parser.parse_args(argv)
    previous = {"report_id":"last-known-good"}
    if args.replace and not args.save_report:
        print("Blocked: --replace requires --save-report", file=sys.stderr)
        return 2
    if args.save_report and args.mode != "plan":
        print("Blocked: only plan mode can save a local report", file=sys.stderr)
        return 2
    try:
        csv_text, fetch, approval, now, seen = scenario(args.scenario)
        if args.scenario == "complete":
            csv_text = load_sample_csv()
        settings = {"site":args.site, "mode":args.mode, "max_pages":args.max_pages}
        if args.mode == "simulate":
            settings.update({"now":now,"approval":approval})
        outcome = build_audit(csv_text, fetch, settings, previous, seen)
        if not isinstance(outcome, dict) or outcome.get("ok") is not True:
            reason = outcome.get("reason","invalid-result") if isinstance(outcome,dict) else "invalid-result"
            print("Blocked: "+str(reason),file=sys.stderr)
            return 2
        if args.save_report:
            destination=write_local_report(outcome["report"],replace=args.replace)
            print("Local report: "+str(destination),file=sys.stderr)
        print(json.dumps(outcome,sort_keys=True,default=str))
        return 0
    except (OSError,ValueError,TypeError,KeyError,NotImplementedError) as exc:
        print("Blocked: "+type(exc).__name__+" "+str(exc),file=sys.stderr)
        return 2

if __name__ == "__main__":
    raise SystemExit(main())

"""Learner-owned acceptance tests: fail at first until build_audit is implemented."""
import copy
import unittest
from datetime import timedelta
from unittest.mock import Mock
from audit_core import build_audit
from fixtures import BASE_CSV, BASE_NOW, APPROVAL, scenario

PREVIOUS = {"report_id":"last-known-good"}
PLAN = {"site":"HQ","mode":"plan","max_pages":3}

class NorthstarAuditTests(unittest.TestCase):
    def audit(self, name="complete", settings=None, previous=None, seen=None):
        csv_text, fetch, approval, now, sample_seen = scenario(name)
        chosen = settings or PLAN
        previous = PREVIOUS if previous is None else previous
        answer = build_audit(csv_text,fetch,chosen,previous, sample_seen if seen is None else seen)
        return answer,fetch,approval,now

    def test_complete_multi_page_report(self):
        actual,fetch,_,_=self.audit()
        self.assertEqual(actual,{'ok':True,'state':'planned',
            'report':{'site':'HQ','total':2,'offline_hosts':['AP-02']},
            'new_keys':[],'previous':PREVIOUS})
        self.assertEqual(fetch.calls,[1,2])

    def test_valid_empty_inventory(self):
        actual,fetch,_,_=self.audit("empty")
        self.assertEqual(actual["report"],{'site':'HQ','total':0,'offline_hosts':[]})
        self.assertEqual(fetch.calls,[1])

    def test_duplicate_csv_blocks_before_fetch(self):
        actual,fetch,_,_=self.audit("duplicate")
        self.assertEqual(actual,{'ok':False,'reason':'duplicate','previous':PREVIOUS})
        self.assertEqual(fetch.calls,[])

    def test_invalid_configuration_fails_closed(self):
        actual,fetch,_,_=self.audit(settings={**PLAN,'max_pages':True})
        self.assertEqual(actual["reason"],"config")
        self.assertEqual(fetch.calls,[])

    def test_unauthorized_does_not_retry(self):
        actual,fetch,_,_=self.audit("unauthorized")
        self.assertEqual(actual["reason"],"http-401")
        self.assertEqual(fetch.calls,[1])

    def test_rate_limited_does_not_retry(self):
        actual,fetch,_,_=self.audit("throttled")
        self.assertEqual(actual["reason"],"http-429")
        self.assertEqual(fetch.calls,[1])

    def test_timeout_recovers_once(self):
        actual,fetch,_,_=self.audit("timeout")
        self.assertTrue(actual["ok"])
        self.assertEqual(fetch.calls,[1,1,2])

    def test_503_exhaustion_preserves_prior(self):
        previous={'report_id':'must-not-overwrite'}
        actual,fetch,_,_=self.audit("unavailable",previous=previous)
        self.assertEqual(actual,{'ok':False,'reason':'http-503','previous':previous})
        self.assertIs(actual["previous"],previous)
        self.assertEqual(fetch.calls,[1,1])

    def test_unsafe_page_and_schema_are_rejected(self):
        bad_page,_,_,_=self.audit("bad-page")
        bad_schema,_,_,_=self.audit("bad-schema")
        self.assertEqual(bad_page["reason"],"page")
        self.assertEqual(bad_schema["reason"],"schema")

    def test_expired_simulation_blocks_before_fetch(self):
        csv_text,fetch,approval,now,_=scenario("expired")
        settings={**PLAN,"mode":"simulate","now":now,"approval":approval}
        actual=build_audit(csv_text,fetch,settings,PREVIOUS,set())
        self.assertEqual(actual["reason"],"approval")
        self.assertEqual(fetch.calls,[])

    def test_simulation_replay_keys_and_input_immutability(self):
        csv_text,fetch,approval,now,seen=scenario("replay")
        settings={**PLAN,"mode":"simulate","now":now,"approval":approval}
        before=(copy.deepcopy(settings),copy.deepcopy(seen),copy.deepcopy(PREVIOUS))
        actual=build_audit(csv_text,fetch,settings,PREVIOUS,seen)
        self.assertEqual(actual["state"],"simulated")
        self.assertEqual(actual["new_keys"],[])
        self.assertEqual((settings,seen,PREVIOUS),before)

    def test_wrong_scope_and_missing_api_host_block(self):
        csv_text,fetch,approval,now,_=scenario()
        wrong={**PLAN,"mode":"simulate","now":now,"approval":{**approval,"site":"OTHER"}}
        self.assertEqual(build_audit(csv_text,fetch,wrong,PREVIOUS,set())["reason"],"approval")
        mismatched=Mock(return_value={'status':200,
                                      'devices':[{'host':'STRANGER','status':'online'}],
                                      'next':None})
        self.assertEqual(build_audit(csv_text,mismatched,PLAN,PREVIOUS,set())["reason"],"mismatch")

if __name__ == "__main__":
    unittest.main()

host,site
SW-01,HQ
AP-02,HQ

# Northstar Fleet Audit — Applied Python Capstone

This is a learner-owned, offline, standard-library-only project. Copy the files
shown on the capstone page into one disposable folder, with `samples/inventory.csv`.
Do not use real customer records, credentials, networks or devices.

## Directory
- audit_core.py — YOUR implementation of build_audit()
- fixtures.py — deterministic, injected HTTP and failure sequences (never network)
- run_audit.py — local argparse entry point; no writes by default
- test_audit.py — 12 independent unittest acceptance checks
- samples/inventory.csv — fictional source data
- reports/ — created only after explicit --save-report

## Start
python -m unittest test_audit.py -v
python run_audit.py --scenario complete
python run_audit.py --scenario unauthorized
python run_audit.py --scenario timeout
python run_audit.py --mode simulate --scenario replay

The tests intentionally FAIL until you implement audit_core.py.
Once the tests pass, also deliberately change an expected value so a test fails;
restore it before calling the project complete.

Plan and simulate modes make no real IT changes. If you explicitly want a LOCAL
JSON artifact after successful plan validation:
python run_audit.py --save-report
python run_audit.py --save-report --replace

The CLI rejects failed/incomplete results and never saves reports in simulate
mode. The local output directory is for training, NOT a production data store.
A file-existence check plus atomic replace is not an inter-process lock; do not
run overlapping writers or treat this project as hardened production software.

## Readiness
Document input contract, supported Python, exact interpreter, scheduling context,
status/exit codes, retry limit, expected host completeness, logs, incident owner,
change approval checks, replay keys and rollback/reconciliation decisions.