import argparse

import io

import json

import subprocess

import sys

import textwrap

import unittest

from contextlib import redirect_stdout

from pathlib import Path

from unittest.mock import patch

import ktx_booking

class FakeTrain:

def __init__(

self,

*,

train_no,

dep_time,

arr_time,

dep_date="20260328",

arr_date="20260328",

run_date="20260328",

train_group="00",

dep_name="서울",

arr_name="부산",

dep_code="0001",

arr_code="0020",

train_type_name="KTX",

has_general_seat=True,

has_special_seat=False,

has_waiting_list=False,

label=None,

):

self.train_no = train_no

self.dep_time = dep_time

self.arr_time = arr_time

self.dep_date = dep_date

self.arr_date = arr_date

self.run_date = run_date

self.train_group = train_group

self.dep_name = dep_name

self.arr_name = arr_name

self.dep_code = dep_code

self.arr_code = arr_code

self.train_type_name = train_type_name

self._has_general_seat = has_general_seat

self._has_special_seat = has_special_seat

self._has_waiting_list = has_waiting_list

self.label = label or train_no

def has_general_seat(self):

return self._has_general_seat

def has_special_seat(self):

return self._has_special_seat

def has_waiting_list(self):

return self._has_waiting_list

def has_general_waiting_list(self):

return self._has_waiting_list

def __str__(self):

return self.label

class FakeReservation:

rsv_id = "320260307102676"

train_no = "009"

train_type_name = "KTX"

dep_name = "서울"

dep_date = "20260328"

dep_time = "090000"

arr_name = "부산"

arr_date = "20260328"

arr_time = "113000"

seat_no_count = 1

price = 59800

buy_limit_date = "20260327"

buy_limit_time = "235900"

journey_no = "001"

journey_cnt = "01"

rsv_chg_no = "00000"

def __str__(self):

return "reservation"

class FakeNCard:

discount_card_no = "1234567890123456"

ticket_kind_name = "N카드"

dep_name = "대전"

arr_name = "서울"

valid = "20260101~20261231"

def __str__(self):

return f"[N카드] {self.dep_name}~{self.arr_name} {self.discount_card_no}"

class FakeClient:

def __init__(self, trains, search_handler=None, ncards=None, ncard_trains=None):

self._trains = trains

self._search_handler = search_handler

self._ncards = ncards or []

self._ncard_trains = ncard_trains or []

self.search_calls = []

self.reserved_train = None

self.reserved_passengers = None

def search_train(self, *args, **kwargs):

self.search_calls.append(kwargs)

if self._search_handler is not None:

return list(self._search_handler(*args, **kwargs))

return list(self._trains)

def reserve(self, train, **kwargs):

self.reserved_train = train

self.reserved_passengers = kwargs.get("passengers")

return FakeReservation()

def owned_ncards(self):

return list(self._ncards)

def search_owned_ncard_trains(self, ncard, **kwargs):

return list(self._ncard_trains)

class KtxBookingTests(unittest.TestCase):

def make_args(self, train_id, ncard_no=None):

return argparse.Namespace(

dep="서울",

arr="부산",

date="20260328",

time="090000",

adults=1,

children=0,

toddlers=0,

seniors=0,

train_id=train_id,

seat_option="general-first",

train_type="ktx",

include_no_seats=False,

include_waiting_list=False,

try_waiting=False,

ncard_no=ncard_no,

)

def test_normalize_train_emits_stable_train_id(self):

train = FakeTrain(train_no="009", dep_time="090000", arr_time="113000")

normalized = ktx_booking.normalize_train(train, index=2)

self.assertIn("train_id", normalized)

train_id = normalized["train_id"]

if not isinstance(train_id, str):

self.fail("train_id should be emitted as a string")

resolved = ktx_booking.find_train_by_id([train], train_id)

self.assertIs(resolved, train)

def test_build_parser_requires_train_id_for_reserve(self):

args = ktx_booking.build_parser().parse_args([

"reserve",

"서울",

"부산",

"20260328",

"090000",

"--train-id",

"ktx:v1:test",

])

self.assertEqual(args.train_id, "ktx:v1:test")

self.assertEqual(args.train_type, "ktx")

def test_build_parser_defaults_search_train_type_to_ktx(self):

args = ktx_booking.build_parser().parse_args([

"search",

"서울",

"부산",

"20260328",

"090000",

])

self.assertEqual(args.train_type, "ktx")

def test_parser_train_type_choices_match_supported_train_types(self):

parser = ktx_booking.build_parser()

for train_type in sorted(ktx_booking.TRAIN_TYPE_MAP):

search_args = parser.parse_args([

"search",

"서울",

"부산",

"20260328",

"090000",

"--train-type",

train_type,

])

reserve_args = parser.parse_args([

"reserve",

"서울",

"부산",

"20260328",

"090000",

"--train-id",

"ktx:v1:test",

"--train-type",

train_type,

])

self.assertEqual(search_args.train_type, train_type)

self.assertEqual(reserve_args.train_type, train_type)

def test_command_search_replays_selected_train_type(self):

selected = FakeTrain(

train_no="2080",

dep_time="155300",

arr_time="170000",

dep_name="남춘천",

arr_name="용산",

train_type_name="ITX-청춘",

)

client = FakeClient([selected])

args = argparse.Namespace(

dep="남춘천",

arr="용산",

date="20260503",

time="150000",

adults=1,

children=0,

toddlers=0,

seniors=0,

limit=5,

train_type="itx-cheongchun",

include_no_seats=False,

include_waiting_list=False,

)

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(io.StringIO()):

ktx_booking.command_search(args)

self.assertEqual(

client.search_calls[-1]["train_type"],

ktx_booking.TRAIN_TYPE_MAP["itx-cheongchun"],

)

def test_command_reserve_targets_exact_train_id_even_if_order_changes(self):

sold_out_first = FakeTrain(

train_no="001",

dep_time="050000",

arr_time="080000",

has_general_seat=False,

label="soldout-first",

)

user_selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="user-selected")

other_train = FakeTrain(train_no="011", dep_time="093000", arr_time="120000", label="other-train")

train_id = ktx_booking.normalize_train(user_selected, index=2)["train_id"]

client = FakeClient([other_train, sold_out_first, user_selected])

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(io.StringIO()):

ktx_booking.command_reserve(self.make_args(train_id))

self.assertIs(client.reserved_train, user_selected)

def test_command_reserve_fails_if_selected_train_is_no_longer_available(self):

user_selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="user-selected")

other_train = FakeTrain(train_no="011", dep_time="093000", arr_time="120000", label="other-train")

train_id = ktx_booking.normalize_train(user_selected, index=2)["train_id"]

client = FakeClient([other_train])

with patch.object(ktx_booking, "build_client", return_value=client):

with self.assertRaises(SystemExit) as exc:

with redirect_stdout(io.StringIO()):

ktx_booking.command_reserve(self.make_args(train_id))

self.assertIn("train_id", str(exc.exception))

def test_command_reserve_replays_selected_train_type(self):

selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")

train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]

client = FakeClient([selected])

args = self.make_args(train_id)

args.train_type = "itx-cheongchun"

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(io.StringIO()):

ktx_booking.command_reserve(args)

self.assertEqual(client.search_calls[-1]["train_type"], ktx_booking.TRAIN_TYPE_MAP["itx-cheongchun"])

self.assertIs(client.reserved_train, selected)

def test_command_reserve_try_waiting_replays_search_with_waiting_list_enabled(self):

waiting_only = FakeTrain(

train_no="003",

dep_time="070000",

arr_time="093000",

has_general_seat=False,

has_special_seat=False,

has_waiting_list=True,

label="waiting-only",

)

train_id = ktx_booking.normalize_train(waiting_only, index=1)["train_id"]

client = FakeClient(

[],

search_handler=lambda *args, **kwargs: [waiting_only] if kwargs.get("include_waiting_list") else [],

)

args = self.make_args(train_id)

args.try_waiting = True

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(io.StringIO()):

ktx_booking.command_reserve(args)

self.assertTrue(client.search_calls)

self.assertTrue(client.search_calls[-1]["include_waiting_list"])

self.assertIs(client.reserved_train, waiting_only)

def test_build_parser_has_ncard_commands(self):

parser = ktx_booking.build_parser()

help_text = parser.format_help()

self.assertIn("ncard-list", help_text)

self.assertIn("ncard-search", help_text)

def test_ncard_search_parser_accepts_ncard_index(self):

args = ktx_booking.build_parser().parse_args([

"ncard-search", "대전", "서울", "20260512", "100000", "--ncard-index", "1",

])

self.assertEqual(args.ncard_index, 1)

self.assertEqual(args.train_type, "ktx")

def test_reserve_parser_accepts_ncard_no(self):

args = ktx_booking.build_parser().parse_args([

"reserve", "대전", "서울", "20260512", "100000",

"--train-id", "ktx:v1:test", "--ncard-no", "1234567890123456",

])

self.assertEqual(args.ncard_no, "1234567890123456")

def test_command_ncard_list_returns_owned_cards(self):

ncard = FakeNCard()

client = FakeClient([], ncards=[ncard])

output = io.StringIO()

with patch.object(ktx_booking, "_NCARD_AVAILABLE", True):

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(output):

ktx_booking.command_ncard_list(argparse.Namespace())

result = json.loads(output.getvalue())

self.assertEqual(result["count"], 1)

self.assertEqual(result["ncards"][0]["index"], 1)

self.assertEqual(result["ncards"][0]["card_no"], "************3456")

self.assertTrue(result["ncards"][0]["card_no_masked"])

self.assertEqual(result["ncards"][0]["dep_name"], ncard.dep_name)

def test_command_ncard_search_returns_trains_with_discount_info(self):

ncard = FakeNCard()

ncard_train = FakeTrain(

train_no="009", dep_time="100000", arr_time="105700",

dep_name="대전", arr_name="서울",

)

ncard_train.price = "9900"

ncard_train.discount_name = "15%할인"

ncard_train.general_remaining_seats = "023"

ncard_train.standing_remaining_seats = None

client = FakeClient([], ncards=[ncard], ncard_trains=[ncard_train])

args = argparse.Namespace(

dep="대전", arr="서울", date="20260512", time="100000",

ncard_index=1, limit=5, train_type="ktx",

)

output = io.StringIO()

with patch.object(ktx_booking, "_NCARD_AVAILABLE", True):

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(output):

ktx_booking.command_ncard_search(args)

result = json.loads(output.getvalue())

self.assertEqual(result["count"], 1)

self.assertEqual(result["trains"][0]["discount_name"], "15%할인")

self.assertEqual(result["trains"][0]["price"], "9900")

self.assertEqual(result["ncard"]["card_no"], "************3456")

def test_command_ncard_search_fails_on_invalid_index(self):

ncard = FakeNCard()

client = FakeClient([], ncards=[ncard])

args = argparse.Namespace(

dep="대전", arr="서울", date="20260512", time="100000",

ncard_index=5, limit=5, train_type="ktx",

)

with patch.object(ktx_booking, "_NCARD_AVAILABLE", True):

with patch.object(ktx_booking, "build_client", return_value=client):

with self.assertRaises(SystemExit):

ktx_booking.command_ncard_search(args)

def test_reserve_parser_accepts_ncard_index(self):

args = ktx_booking.build_parser().parse_args([

"reserve", "대전", "서울", "20260512", "100000",

"--train-id", "ktx:v1:test", "--ncard-index", "1",

])

self.assertEqual(args.ncard_index, 1)

def test_command_ncard_search_fails_on_zero_index(self):

ncard = FakeNCard()

client = FakeClient([], ncards=[ncard])

args = argparse.Namespace(

dep="대전", arr="서울", date="20260512", time="100000",

ncard_index=0, limit=5, train_type="ktx",

)

with patch.object(ktx_booking, "_NCARD_AVAILABLE", True):

with patch.object(ktx_booking, "build_client", return_value=client):

with self.assertRaises(SystemExit) as exc:

ktx_booking.command_ncard_search(args)

self.assertIn("ncard-index", str(exc.exception))

def test_command_ncard_list_requires_ncard_package(self):

with patch.object(ktx_booking, "_NCARD_AVAILABLE", False):

with self.assertRaises(SystemExit) as exc:

ktx_booking.command_ncard_list(argparse.Namespace())

self.assertIn("korail2-ncard", str(exc.exception))

def test_command_reserve_with_ncard_no_requires_ncard_package(self):

selected = FakeTrain(train_no="009", dep_time="100000", arr_time="105700",

dep_name="대전", arr_name="서울")

train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]

args = self.make_args(train_id, ncard_no="1234567890123456")

args.ncard_index = None

client = FakeClient([selected])

with patch.object(ktx_booking, "_NCARD_AVAILABLE", False):

with patch.object(ktx_booking, "build_client", return_value=client):

with self.assertRaises(SystemExit) as exc:

ktx_booking.command_reserve(args)

self.assertIn("korail2-ncard", str(exc.exception))

def test_command_reserve_with_ncard_no_uses_ncard_passenger(self):

selected = FakeTrain(train_no="009", dep_time="100000", arr_time="105700",

dep_name="대전", arr_name="서울")

train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]

client = FakeClient([selected])

args = self.make_args(train_id, ncard_no="1234567890123456")

args.ncard_index = None

with patch.object(ktx_booking, "_NCARD_AVAILABLE", True):

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(io.StringIO()):

ktx_booking.command_reserve(args)

self.assertIsNotNone(client.reserved_passengers)

self.assertEqual(len(client.reserved_passengers), 1)

self.assertIsInstance(client.reserved_passengers[0], ktx_booking.NCardPassenger)

def test_command_reserve_with_ncard_index_uses_owned_card_without_exposing_full_number(self):

selected = FakeTrain(train_no="009", dep_time="100000", arr_time="105700",

dep_name="대전", arr_name="서울")

train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]

client = FakeClient([selected], ncards=[FakeNCard()])

args = self.make_args(train_id)

args.ncard_index = 1

with patch.object(ktx_booking, "_NCARD_AVAILABLE", True):

with patch.object(ktx_booking, "build_client", return_value=client):

with redirect_stdout(io.StringIO()):

ktx_booking.command_reserve(args)

self.assertEqual(client.reserved_passengers[0].card_no, "1234567890123456")

def test_ncard_unavailable_error_message_is_shared(self):

with patch.object(ktx_booking, "_NCARD_AVAILABLE", False):

with self.assertRaises(SystemExit) as exc:

ktx_booking.ensure_ncard_available()

self.assertIn("korail2-ncard", str(exc.exception))

class FallbackImportTests(unittest.TestCase):

def test_module_imports_when_korail2_is_missing(self):

script_dir = Path(__file__).resolve().parent

helper = textwrap.dedent(

"""

import importlib

import sys

sys.modules["korail2"] = None

sys.modules.pop("ktx_booking", None)

module = importlib.import_module("ktx_booking")

assert module._KORAIL_IMPORT_ERROR is not None, "expected fallback path"

assert module.TRAIN_TYPE_MAP["ktx"] == "100"

assert module.TRAIN_TYPE_MAP["itx-cheongchun"] == "104"

assert module.TRAIN_TYPE_MAP["itx-saemaeul"] == "101"

assert module.TRAIN_TYPE_MAP["mugunghwa"] == "102"

assert module.TRAIN_TYPE_MAP["nuriro"] == "102"

assert module.TRAIN_TYPE_MAP["tonggeun"] == "103"

assert module.TRAIN_TYPE_MAP["airport"] == "105"

assert module.TRAIN_TYPE_MAP["all"] == "109"

print("ok")

"""

).strip()

env = {

"PYTHONPATH": str(script_dir),

"PYTHONNOUSERSITE": "1",

"PATH": "",

}

result = subprocess.run(

[sys.executable, "-S", "-c", helper],

capture_output=True,

text=True,

env=env,

check=False,

)

self.assertEqual(result.returncode, 0, msg=result.stderr)

self.assertIn("ok", result.stdout)

def test_help_works_when_korail2_is_missing(self):

script_dir = Path(__file__).resolve().parent

helper = textwrap.dedent(

"""

import importlib

import sys

sys.modules["korail2"] = None

sys.modules.pop("ktx_booking", None)

module = importlib.import_module("ktx_booking")

parser = module.build_parser()

help_text = parser.format_help()

assert "search" in help_text

assert "reserve" in help_text

print("ok")

"""

).strip()

env = {

"PYTHONPATH": str(script_dir),

"PYTHONNOUSERSITE": "1",

"PATH": "",

}

result = subprocess.run(

[sys.executable, "-S", "-c", helper],

capture_output=True,

text=True,

env=env,

check=False,

)

self.assertEqual(result.returncode, 0, msg=result.stderr)

self.assertIn("ok", result.stdout)

if __name__ == "__main__":

unittest.main()