flowmark 0.3.1

A Markdown auto-formatter for clean diffs and semantic line breaks
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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

#!/usr/bin/env python3
"""Tests for scripts/resolve_crate_metadata.py."""

from __future__ import annotations

import http.server
import subprocess
import sys
import tempfile
import textwrap
import threading
import unittest
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT_PATH = REPO_ROOT / "scripts" / "resolve_crate_metadata.py"


def _start_stub_server(status_code: int, expected_path: str):
    class Handler(http.server.BaseHTTPRequestHandler):
        def do_GET(self) -> None:  # noqa: N802 (stdlib interface)
            code = status_code if self.path == expected_path else 400
            self.send_response(code)
            self.end_headers()
            self.wfile.write(b"{}")

        def log_message(self, _format: str, *_args) -> None:  # noqa: A003
            return

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


class ResolveCrateMetadataTests(unittest.TestCase):
    maxDiff = None

    def _run_script(self, *, cargo_toml: str, status_code: int) -> tuple[subprocess.CompletedProcess[str], str]:
        expected_path = "/api/v1/crates/flowmark-rs/1.2.3"
        server, thread = _start_stub_server(status_code, expected_path)
        output_contents = ""

        try:
            with tempfile.TemporaryDirectory() as temp_dir:
                temp_path = Path(temp_dir)
                manifest_path = temp_path / "Cargo.toml"
                output_path = temp_path / "github_output.txt"
                manifest_path.write_text(cargo_toml, encoding="utf-8")

                base_url = f"http://127.0.0.1:{server.server_port}/api/v1/crates"
                result = subprocess.run(
                    [
                        sys.executable,
                        str(SCRIPT_PATH),
                        "--manifest-path",
                        str(manifest_path),
                        "--registry-url",
                        base_url,
                        "--github-output",
                        str(output_path),
                    ],
                    capture_output=True,
                    text=True,
                )

                if output_path.exists():
                    output_contents = output_path.read_text(encoding="utf-8")
        finally:
            server.shutdown()
            thread.join(timeout=5)
            server.server_close()

        return result, output_contents

    def test_detects_already_published_false_when_version_missing(self) -> None:
        cargo_toml = textwrap.dedent(
            """
            [package]
            name = "flowmark-rs"
            version = "1.2.3"
            """
        )
        result, output = self._run_script(cargo_toml=cargo_toml, status_code=404)
        self.assertEqual(result.returncode, 0, msg=result.stderr)
        self.assertIn("crate_name=flowmark-rs", output)
        self.assertIn("crate_version=1.2.3", output)
        self.assertIn("already_published=false", output)

    def test_detects_already_published_true_when_version_exists(self) -> None:
        cargo_toml = textwrap.dedent(
            """
            [package]
            name = "flowmark-rs"
            version = "1.2.3"
            """
        )
        result, output = self._run_script(cargo_toml=cargo_toml, status_code=200)
        self.assertEqual(result.returncode, 0, msg=result.stderr)
        self.assertIn("already_published=true", output)

    def test_fails_for_manifest_without_package_version(self) -> None:
        cargo_toml = textwrap.dedent(
            """
            [package]
            name = "flowmark-rs"
            """
        )
        result, _output = self._run_script(cargo_toml=cargo_toml, status_code=404)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("Failed to resolve crate name/version", result.stderr)

    def test_fails_for_registry_server_errors(self) -> None:
        cargo_toml = textwrap.dedent(
            """
            [package]
            name = "flowmark-rs"
            version = "1.2.3"
            """
        )
        result, _output = self._run_script(cargo_toml=cargo_toml, status_code=500)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("Unexpected response from registry", result.stderr)


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