221 lines
8.4 KiB
Python
221 lines
8.4 KiB
Python
"""Tests for the BLM CLI subprocess wrapper and output parsers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from link_building_workflow import BLMConfig
|
|
from link_building_workflow.blm import (
|
|
build_ingest_args,
|
|
parse_generate_output,
|
|
parse_ingest_output,
|
|
run_blm_command,
|
|
)
|
|
|
|
|
|
class TestBuildIngestArgs:
|
|
def test_required_args_only(self):
|
|
args = build_ingest_args(xlsx_path="/tmp/f.xlsx", project_name="P")
|
|
assert args == ["ingest-cora", "-f", "/tmp/f.xlsx", "-n", "P"]
|
|
|
|
def test_with_money_site_url(self):
|
|
args = build_ingest_args(
|
|
xlsx_path="/tmp/f.xlsx",
|
|
project_name="P",
|
|
money_site_url="https://example.com",
|
|
)
|
|
assert "-m" in args
|
|
i = args.index("-m")
|
|
assert args[i + 1] == "https://example.com"
|
|
|
|
def test_branded_plus_ratio_default_omitted(self):
|
|
args = build_ingest_args(
|
|
xlsx_path="/tmp/f.xlsx", project_name="P", branded_plus_ratio=0.7
|
|
)
|
|
assert "-bp" not in args
|
|
|
|
def test_branded_plus_ratio_custom_included(self):
|
|
args = build_ingest_args(
|
|
xlsx_path="/tmp/f.xlsx", project_name="P", branded_plus_ratio=0.8
|
|
)
|
|
assert "-bp" in args
|
|
assert args[args.index("-bp") + 1] == "0.8"
|
|
|
|
def test_custom_anchors(self):
|
|
args = build_ingest_args(
|
|
xlsx_path="/tmp/f.xlsx",
|
|
project_name="P",
|
|
custom_anchors="a1,a2",
|
|
)
|
|
assert "-a" in args
|
|
assert args[args.index("-a") + 1] == "a1,a2"
|
|
|
|
def test_cli_flags_split_on_whitespace(self):
|
|
args = build_ingest_args(
|
|
xlsx_path="/tmp/f.xlsx",
|
|
project_name="P",
|
|
cli_flags="--foo --bar baz",
|
|
)
|
|
assert "--foo" in args
|
|
assert "--bar" in args
|
|
assert "baz" in args
|
|
|
|
def test_cli_flags_empty_string_no_extra_args(self):
|
|
args = build_ingest_args(
|
|
xlsx_path="/tmp/f.xlsx", project_name="P", cli_flags=""
|
|
)
|
|
assert args == ["ingest-cora", "-f", "/tmp/f.xlsx", "-n", "P"]
|
|
|
|
|
|
class TestParseIngestOutput:
|
|
def test_full_success(self, ingest_success_stdout):
|
|
result = parse_ingest_output(ingest_success_stdout)
|
|
assert result.project_id == "42"
|
|
assert result.project_name == "Test Project"
|
|
assert result.main_keyword == "precision cnc machining"
|
|
assert result.job_file == "jobs/test-project.json"
|
|
assert result.success is True
|
|
|
|
def test_missing_project_line(self):
|
|
stdout = "Job file created: jobs/x.json\n"
|
|
result = parse_ingest_output(stdout)
|
|
assert result.project_id == ""
|
|
assert result.project_name == ""
|
|
assert result.success is False # no project_id
|
|
|
|
def test_missing_job_line(self):
|
|
stdout = "Success: Project 'X' created (ID: 1)\n"
|
|
result = parse_ingest_output(stdout)
|
|
assert result.project_id == "1"
|
|
assert result.job_file == ""
|
|
assert result.success is False # no job_file
|
|
|
|
def test_empty_stdout(self):
|
|
result = parse_ingest_output("")
|
|
assert result.project_id == ""
|
|
assert result.job_file == ""
|
|
assert result.success is False
|
|
|
|
def test_ignores_noise(self):
|
|
stdout = (
|
|
"Some random banner\n"
|
|
"DEBUG: lots of stuff\n"
|
|
"Success: Project 'Foo Bar' created (ID: 99)\n"
|
|
"WARNING: meaningless\n"
|
|
"Main Keyword: foo bar\n"
|
|
"Job file created: jobs/foo-bar.json\n"
|
|
"Done.\n"
|
|
)
|
|
result = parse_ingest_output(stdout)
|
|
assert result.project_id == "99"
|
|
assert result.project_name == "Foo Bar"
|
|
assert result.main_keyword == "foo bar"
|
|
assert result.job_file == "jobs/foo-bar.json"
|
|
|
|
def test_whitespace_around_job_file(self):
|
|
stdout = "Job file created: jobs/x.json \n"
|
|
result = parse_ingest_output(stdout)
|
|
assert result.job_file == "jobs/x.json"
|
|
|
|
|
|
class TestParseGenerateOutput:
|
|
def test_success_with_move(self, generate_success_stdout):
|
|
result = parse_generate_output(generate_success_stdout)
|
|
assert result.success is True
|
|
assert result.job_moved_to == "jobs/done/test-project.json"
|
|
assert "Job file moved to" in result.raw_output
|
|
|
|
def test_no_move_line(self):
|
|
stdout = "Generating backlinks...\nSome error occurred.\n"
|
|
result = parse_generate_output(stdout)
|
|
assert result.success is False
|
|
assert result.job_moved_to == ""
|
|
assert result.raw_output == stdout
|
|
|
|
def test_empty_stdout(self):
|
|
result = parse_generate_output("")
|
|
assert result.success is False
|
|
assert result.job_moved_to == ""
|
|
|
|
|
|
class TestRunBlmCommand:
|
|
def test_passes_cwd_and_interpreter(self, blm_config: BLMConfig):
|
|
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
run_blm_command(["ingest-cora", "-f", "x.xlsx"], blm_config)
|
|
call = mock_run.call_args
|
|
cmd = call[0][0]
|
|
assert cmd[0] == "python"
|
|
assert cmd[1] == "main.py"
|
|
assert "ingest-cora" in cmd
|
|
assert call[1]["cwd"] == blm_config.blm_dir
|
|
|
|
def test_injects_credentials(self, blm_config: BLMConfig):
|
|
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
run_blm_command(["ingest-cora"], blm_config)
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "-u" in cmd
|
|
assert cmd[cmd.index("-u") + 1] == "testuser"
|
|
assert "-p" in cmd
|
|
assert cmd[cmd.index("-p") + 1] == "testpass"
|
|
|
|
def test_does_not_duplicate_user_flag(self, blm_config: BLMConfig):
|
|
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
run_blm_command(["ingest-cora", "-u", "other"], blm_config)
|
|
cmd = mock_run.call_args[0][0]
|
|
# -u should appear once, with the caller's value preserved
|
|
assert cmd.count("-u") == 1
|
|
assert cmd[cmd.index("-u") + 1] == "other"
|
|
|
|
def test_does_not_duplicate_password_flag(self, blm_config: BLMConfig):
|
|
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
run_blm_command(["ingest-cora", "-p", "otherpw"], blm_config)
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd.count("-p") == 1
|
|
assert cmd[cmd.index("-p") + 1] == "otherpw"
|
|
|
|
def test_skips_credentials_when_not_configured(self, tmp_path: Path):
|
|
blm_dir = tmp_path / "blm"
|
|
blm_dir.mkdir()
|
|
config = BLMConfig(blm_dir=str(blm_dir)) # no user/pass
|
|
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
run_blm_command(["ingest-cora"], config)
|
|
cmd = mock_run.call_args[0][0]
|
|
assert "-u" not in cmd
|
|
assert "-p" not in cmd
|
|
|
|
def test_raises_on_missing_blm_dir(self, tmp_path: Path):
|
|
config = BLMConfig(blm_dir=str(tmp_path / "nope"))
|
|
with pytest.raises(FileNotFoundError):
|
|
run_blm_command(["ingest-cora"], config)
|
|
|
|
def test_passes_timeout(self, blm_config: BLMConfig):
|
|
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
run_blm_command(["ingest-cora"], blm_config)
|
|
assert mock_run.call_args[1]["timeout"] == blm_config.timeout_seconds
|
|
|
|
def test_propagates_timeout_expired(self, blm_config: BLMConfig):
|
|
with patch(
|
|
"subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(cmd="python", timeout=300),
|
|
), pytest.raises(subprocess.TimeoutExpired):
|
|
run_blm_command(["ingest-cora"], blm_config)
|
|
|
|
def test_custom_python_exe(self, tmp_path: Path):
|
|
blm_dir = tmp_path / "blm"
|
|
blm_dir.mkdir()
|
|
config = BLMConfig(blm_dir=str(blm_dir), python_exe="/opt/venv/bin/python")
|
|
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
run_blm_command(["ingest-cora"], config)
|
|
assert mock_run.call_args[0][0][0] == "/opt/venv/bin/python"
|