mirror of
https://github.com/NomaDamas/k-skill.git
synced 2026-06-24 02:04:11 +00:00
Merge remote-tracking branch 'origin/pr/298' into ulw-merge-seven-prs
This commit is contained in:
commit
9dc3577742
6 changed files with 1307 additions and 17 deletions
|
|
@ -23,7 +23,7 @@ Claude Code, Codex, OpenCode, OpenClaw/ClawHub 등 각종 코딩 에이전트
|
|||
| 할 수 있는 일 | 스킬 이름 | 설명 | 사용자 로그인 | 문서 |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| SRT 예매 | `srt-booking` | SRT 열차 조회, 예약, 예약 확인, 취소 | 필요 | [SRT 예매 가이드](docs/features/srt-booking.md) |
|
||||
| KTX 예매 | `ktx-booking` | KTX/Korail 열차 조회, 예약, 예약 확인, 취소 | 필요 | [KTX 예매 가이드](docs/features/ktx-booking.md) |
|
||||
| KTX 예매 | `ktx-booking` | KTX/Korail 열차 조회, 호차별 좌석번호·콘센트 좌석 확인, 예약, 예약 확인, 취소 | 필요 | [KTX 예매 가이드](docs/features/ktx-booking.md) |
|
||||
| 고속버스 예매 | `express-bus-booking` | KOBUS 고속버스 배차·좌석·요금 조회와 결제 직전 handoff(결제는 수동) | 불필요 | [고속버스 예매 가이드](docs/features/express-bus-booking.md) |
|
||||
| 시외버스 예매 | `intercity-bus-booking` | 티머니 시외버스 배차·좌석·요금 조회와 결제 직전 handoff(결제는 수동) | 불필요 | [시외버스 예매 가이드](docs/features/intercity-bus-booking.md) |
|
||||
| 자연휴양림 빈 객실 조회 | `foresttrip-vacancy` | 공식 숲나들e 자연휴양림 예약 가능 객실 조회 자동화 (예약/결제 제외) | 필요 | [자연휴양림 빈 객실 조회 가이드](docs/features/foresttrip-vacancy.md) |
|
||||
|
|
|
|||
|
|
@ -4,6 +4,8 @@
|
|||
|
||||
- KTX/Korail 열차 조회
|
||||
- 좌석 가능 여부 확인
|
||||
- 호차별 남은 좌석번호 확인
|
||||
- 콘센트 꿀팁 좌석 필터링
|
||||
- 예약 진행
|
||||
- 예약 내역 확인
|
||||
- 예약 취소
|
||||
|
|
@ -35,6 +37,7 @@
|
|||
- 희망 시작 시각: `HHMMSS`
|
||||
- 인원 수와 승객 유형
|
||||
- 좌석 선호
|
||||
- 좌석 상세 조건: 객실 등급, 호차 번호, 남은 좌석만 보기, 콘센트 좌석 우선
|
||||
- 조회 결과에서 복사한 `train_id`
|
||||
|
||||
## 왜 helper 를 쓰는가
|
||||
|
|
@ -54,8 +57,9 @@
|
|||
2. `KSKILL_KTX_ID`, `KSKILL_KTX_PASSWORD` 가 없으면 credential resolution order에 따라 확보한다.
|
||||
3. helper 로 먼저 열차를 조회한다.
|
||||
4. 후보 열차의 `index`, `train_id`, 출발/도착 시각, KTX 여부, 좌석 여부를 보여준다.
|
||||
5. 대상 열차가 명확할 때만 예약한다.
|
||||
6. 예약 확인/취소는 대상 예약을 다시 식별한 뒤 진행한다.
|
||||
5. 사용자가 좌석번호, 호차별 잔여석, 콘센트 꿀팁 좌석을 물으면 `seats` 로 상세 좌석을 먼저 확인한다.
|
||||
6. 대상 열차가 명확할 때만 예약한다.
|
||||
7. 예약 확인/취소는 대상 예약을 다시 식별한 뒤 진행한다.
|
||||
|
||||
## 예시
|
||||
|
||||
|
|
@ -69,6 +73,43 @@ python3 scripts/ktx_booking.py search 서울 부산 20260328 090000 --limit 5
|
|||
|
||||
응답 JSON 의 `train_id` 는 검색 시점의 정확한 열차를 가리키는 stable selector 다. 예약할 때는 이 값을 그대로 복사해서 쓴다. 같은 열차가 더 이상 조회되지 않으면 helper 가 실패하고 새로 조회하게 만든다.
|
||||
|
||||
상세 좌석 확인:
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id>
|
||||
```
|
||||
|
||||
남은 좌석번호만 확인:
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only
|
||||
```
|
||||
|
||||
특정 호차를 지정하지 않으면 `seats` 는 승강장 이동 거리가 짧은 가운데 호차부터 탐색한다. 각 호차 안에서는 콘센트 힌트가 있는 좌석을 먼저, 같은 조건에서는 순방향 좌석을 먼저 반환한다.
|
||||
|
||||
특정 호차의 남은 좌석만 확인:
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --car-no 5 --available-only
|
||||
```
|
||||
|
||||
콘센트 꿀팁 좌석부터 확인:
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only --power-only
|
||||
```
|
||||
|
||||
특실 좌석을 확인하려면 `--room special`, KTX 외 열차를 조회했다면 `search` 와 같은 `--train-type` 을 함께 넘긴다.
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 남춘천 용산 20260503 150000 \
|
||||
--train-id <train_id> \
|
||||
--train-type itx-cheongchun \
|
||||
--available-only
|
||||
```
|
||||
|
||||
`seats` 응답은 호차별 `remaining_seats`, `available_seats`, 좌석별 순방향/역방향, 창측/내측, 좌석 종류, 문 근처 여부, 콘센트 힌트를 JSON 으로 반환한다. 이 단계는 좌석을 선택하거나 선점하지 않고, 예약 전 확인만 한다.
|
||||
|
||||
예약:
|
||||
|
||||
```bash
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
---
|
||||
name: ktx-booking
|
||||
description: Search, reserve, inspect, and cancel KTX or Korail tickets in Korea with the korail2 + pycryptodome Python packages. Use when the user asks for KTX seats, Korail bookings, train changes, or reservation status.
|
||||
description: Search, reserve, inspect, and cancel KTX or Korail tickets in Korea with the korail2 + pycryptodome Python packages. Use when the user asks for KTX seats, Korail bookings, train changes, reservation status, remaining seat numbers, car-by-car seats, or power-outlet/good-seat tips.
|
||||
license: MIT
|
||||
metadata:
|
||||
category: travel
|
||||
|
|
@ -12,7 +12,7 @@ metadata:
|
|||
|
||||
## What this skill does
|
||||
|
||||
`korail2` 위에 `scripts/ktx_booking.py` helper 를 얹어 KTX/Korail 조회, 예약, 예약 확인, 취소를 처리한다.
|
||||
`korail2` 위에 `scripts/ktx_booking.py` helper 를 얹어 KTX/Korail 조회, 호차별 좌석번호 확인, 예약, 예약 확인, 취소를 처리한다.
|
||||
|
||||
최근 Korail 앱의 Dynapath anti-bot 체크 때문에 원본 `korail2` 0.4.0 예제만으로는 `MACRO ERROR` 가 날 수 있다. 이 스킬은 helper 가 `x-dynapath-m-token`, `Sid`, 최신 app version(`250601002`)을 붙여 실제 예매 흐름을 복구하는 것을 전제로 한다.
|
||||
|
||||
|
|
@ -22,6 +22,10 @@ metadata:
|
|||
- "코레일 예약 확인해줘"
|
||||
- "KTX 취소해줘"
|
||||
- "오전 9시 이후 KTX 중 제일 빠른 거 잡아줘"
|
||||
- "KTX 남은 좌석 번호 확인해줘"
|
||||
- "이 열차 콘센트 있는 꿀팁 좌석부터 보여줘"
|
||||
- "KTX 5호차 남은 자리만 봐줘"
|
||||
- "예약하기 전에 호차별 좌석 확인해줘"
|
||||
- "N카드로 할인 열차 찾아줘"
|
||||
- "내 N카드 목록 보여줘"
|
||||
- "N카드 할인 적용해서 예약해줘"
|
||||
|
|
@ -59,6 +63,7 @@ metadata:
|
|||
- 희망 시작 시각: `HHMMSS`
|
||||
- 인원 수와 승객 유형
|
||||
- 좌석 선호
|
||||
- 좌석 상세 조건: 객실 등급, 호차 번호, 남은 좌석만 보기, 콘센트 꿀팁 좌석 우선
|
||||
- 조회 결과에서 복사한 `train_id`
|
||||
|
||||
## Workflow
|
||||
|
|
@ -108,7 +113,62 @@ python3 scripts/ktx_booking.py search 남춘천 용산 20260503 150000 --train-t
|
|||
- 일반실/특실 가능 여부
|
||||
- 예약 대기 가능 여부
|
||||
|
||||
### 4. Reserve only after the target train is unambiguous
|
||||
### 4. Inspect detailed seats when the user asks for good seats
|
||||
|
||||
`search` 의 좌석 가능 여부는 열차 단위 플래그다. 사용자가 "남은 좌석 번호", "호차별 좌석", "콘센트", "꿀팁 좌석", "창측/순방향 자리", "예약 전에 자리 확인"처럼 구체적인 좌석을 물으면 예약 전에 `seats` 를 호출한다.
|
||||
|
||||
기본 상세 좌석 조회:
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id>
|
||||
```
|
||||
|
||||
일반실/특실은 `--room` 으로 나눈다.
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --room special
|
||||
```
|
||||
|
||||
남은 좌석번호만 보고 싶으면 `--available-only` 를 쓴다.
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only
|
||||
```
|
||||
|
||||
특정 호차를 지정하지 않으면 `seats` 는 승강장 이동 거리가 짧은 가운데 호차부터 탐색한다. 각 호차 안의 좌석은 콘센트 힌트가 있는 좌석(`direct`, `adjacent`)을 먼저, 같은 조건에서는 순방향 좌석을 먼저 보여준다.
|
||||
|
||||
특정 호차만 확인하려면 `--car-no` 를 쓴다.
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --car-no 5 --available-only
|
||||
```
|
||||
|
||||
콘센트 꿀팁 자리부터 확인하려면 `--power-only` 를 붙인다. 응답의 `power_outlet` 은 `direct`, `adjacent`, `none` 중 하나다.
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 서울 부산 20260328 090000 --train-id <train_id> --available-only --power-only
|
||||
```
|
||||
|
||||
`seats` 도 `search` 와 같은 `--train-type` 을 넘겨야 한다. ITX-청춘 등 KTX 외 열차를 조회했다면 상세 좌석 조회에도 같은 값을 사용한다.
|
||||
|
||||
```bash
|
||||
python3 scripts/ktx_booking.py seats 남춘천 용산 20260503 150000 \
|
||||
--train-id <train_id> \
|
||||
--train-type itx-cheongchun \
|
||||
--available-only
|
||||
```
|
||||
|
||||
상세 좌석 응답을 보여줄 때는 사용자 의도에 맞춰 아래를 우선 요약한다.
|
||||
|
||||
- 호차별 `remaining_seats`, `available_seat_count`
|
||||
- 남은 좌석 번호 (`available_seats`)
|
||||
- 좌석별 `direction`, `position`, `seat_type`
|
||||
- 콘센트 힌트 (`power_outlet`)
|
||||
- 문 근처 여부 (`near_door`)
|
||||
|
||||
이 기능은 좌석을 선택/선점하지 않는다. 실제 예약은 다음 단계의 `reserve` 로만 진행한다.
|
||||
|
||||
### 5. Reserve only after the target train is unambiguous
|
||||
|
||||
조회 결과의 `train_id` 를 고른 뒤에만 예약한다. 이 값은 helper 가 열차 번호/운행일/시각/역 코드를 묶어 만든 stable selector 이므로, 재조회 시 같은 열차가 아직 있으면 그대로 잡고 없으면 실패한다.
|
||||
|
||||
|
|
@ -125,7 +185,7 @@ python3 scripts/ktx_booking.py reserve 남춘천 용산 20260503 150000 --train-
|
|||
응답에는 예약번호, 운임, 구입기한이 포함된다. **결제는 자동화하지 않는다.**
|
||||
좌석이 없을 때는 조회 단계에서 `--include-waiting-list` 를 켜고 예약 단계에서 `--try-waiting` 으로 예약 대기까지 시도할 수 있다.
|
||||
|
||||
### 4-1. N-card discounted reservation
|
||||
### 5-1. N-card discounted reservation
|
||||
|
||||
N카드 할인을 적용하려면 먼저 보유 N카드 목록을 조회해 카드 번호를 확인한다.
|
||||
|
||||
|
|
@ -151,7 +211,7 @@ python3 scripts/ktx_booking.py reserve 대전 서울 20260512 100000 \
|
|||
|
||||
N카드 기능은 `korail2-ncard` 패키지가 필요하다. 없으면 해당 커맨드 실행 시 설치 안내가 출력된다.
|
||||
|
||||
### 5. Inspect or cancel
|
||||
### 6. Inspect or cancel
|
||||
|
||||
취소는 대상 예약을 다시 조회해 식별한 뒤에만 진행한다.
|
||||
|
||||
|
|
@ -166,6 +226,7 @@ python3 scripts/ktx_booking.py cancel <reservation_id>
|
|||
## Done when
|
||||
|
||||
- 조회면 열차 후보가 정리되어 있다
|
||||
- 좌석 상세 확인이면 호차별 남은 좌석번호와 필요한 꿀팁 조건이 정리되어 있다
|
||||
- 예약이면 예약 결과와 제한 시간이 확인되어 있다
|
||||
- 취소면 어떤 예약을 취소했는지 남아 있다
|
||||
|
||||
|
|
|
|||
|
|
@ -103,7 +103,7 @@ except ModuleNotFoundError as exc:
|
|||
|
||||
class _FallbackKorailModule:
|
||||
EMAIL_REGEX = re.compile(r".+@.+")
|
||||
PHONE_NUMBER_REGEX = re.compile(r"^\d+$")
|
||||
PHONE_NUMBER_REGEX = re.compile(r"(\d{3})-(\d{3,4})-(\d{4})")
|
||||
|
||||
korail_mod = _FallbackKorailModule()
|
||||
else:
|
||||
|
|
@ -127,11 +127,15 @@ DEFAULT_USER_AGENT = "Dalvik/2.1.0 (Linux; U; Android 13; SM-S928N Build/UP1A.23
|
|||
DYNAPATH_PATHS = [
|
||||
"/classes/com.korail.mobile.certification.TicketReservation",
|
||||
"/classes/com.korail.mobile.nonMember.NonMemTicket",
|
||||
"/classes/com.korail.mobile.research.TrainResearch",
|
||||
"/classes/com.korail.mobile.research.ResidualSeatsResearch.do",
|
||||
"/classes/com.korail.mobile.seatMovie.ScheduleView",
|
||||
"/classes/com.korail.mobile.seatMovie.ScheduleViewSpecial",
|
||||
"/classes/com.korail.mobile.trn.prcFare.do",
|
||||
"/classes/com.korail.mobile.login.Login",
|
||||
]
|
||||
KORAIL_CARS_INFO = "https://smart.letskorail.com:443/classes/com.korail.mobile.research.TrainResearch"
|
||||
KORAIL_CAR_DETAIL = "https://smart.letskorail.com:443/classes/com.korail.mobile.research.ResidualSeatsResearch.do"
|
||||
RESERVE_OPTION_MAP = {
|
||||
"general-first": ReserveOption.GENERAL_FIRST,
|
||||
"general-only": ReserveOption.GENERAL_ONLY,
|
||||
|
|
@ -163,6 +167,43 @@ TRAIN_ID_FIELDS = (
|
|||
"arr_code",
|
||||
)
|
||||
|
||||
PHONE_NUMBER_DIGITS_REGEX = re.compile(r"^01\d{8,9}$")
|
||||
ROOM_CLASS_MAP = {
|
||||
"general": "1",
|
||||
"special": "2",
|
||||
}
|
||||
ROOM_CLASS_NAME = {
|
||||
"1": "일반실",
|
||||
"2": "특실",
|
||||
}
|
||||
SEAT_DIRECTION_NAME = {
|
||||
"009": "순방향",
|
||||
"010": "역방향",
|
||||
}
|
||||
SEAT_POSITION_NAME = {
|
||||
"011": "1인",
|
||||
"012": "창측",
|
||||
"013": "내측",
|
||||
}
|
||||
SEAT_TYPE_NAME = {
|
||||
"015": "일반석",
|
||||
"018": "2층석",
|
||||
"019": "유아동반석",
|
||||
"021": "휠체어석",
|
||||
"023": "4인동반석",
|
||||
"027": "4인석",
|
||||
"028": "전동휠체어석",
|
||||
"032": "자전거",
|
||||
"052": "대피도우미",
|
||||
}
|
||||
POWER_OUTLET_ROWS = {1, 3, 5, 7, 10, 12, 14, 15}
|
||||
POWER_OUTLET_DIRECT_COLUMNS = {"A", "D"}
|
||||
POWER_OUTLET_ADJACENT_COLUMNS = {"B", "C"}
|
||||
|
||||
|
||||
def is_phone_login_id(korail_id: str) -> bool:
|
||||
return bool(korail_mod.PHONE_NUMBER_REGEX.fullmatch(korail_id) or PHONE_NUMBER_DIGITS_REGEX.fullmatch(korail_id))
|
||||
|
||||
|
||||
def ensure_runtime_dependencies() -> None:
|
||||
missing: list[str] = []
|
||||
|
|
@ -334,7 +375,7 @@ class PatchedKorail(Korail):
|
|||
|
||||
if korail_mod.EMAIL_REGEX.match(korail_id):
|
||||
input_flag = "5"
|
||||
elif korail_mod.PHONE_NUMBER_REGEX.match(korail_id):
|
||||
elif is_phone_login_id(korail_id):
|
||||
input_flag = "4"
|
||||
else:
|
||||
input_flag = "2"
|
||||
|
|
@ -364,7 +405,7 @@ class PatchedKorail(Korail):
|
|||
self.logined = False
|
||||
return False
|
||||
|
||||
def search_train(
|
||||
def search_train_details(
|
||||
self,
|
||||
dep: str,
|
||||
arr: str,
|
||||
|
|
@ -424,17 +465,98 @@ class PatchedKorail(Korail):
|
|||
response = self._session.post(korail_mod.KORAIL_SEARCH_SCHEDULE, params=payload, headers=headers)
|
||||
data = json.loads(response.text)
|
||||
if self._result_check(data):
|
||||
trains = [korail_mod.Train(info) for info in data["trn_infos"]["trn_info"]]
|
||||
trains = [train for train in trains if train.dep_name == dep and train.arr_name == arr]
|
||||
train_infos = data["trn_infos"]["trn_info"]
|
||||
if isinstance(train_infos, dict):
|
||||
train_infos = [train_infos]
|
||||
details = [(korail_mod.Train(info), info) for info in train_infos]
|
||||
details = [(train, info) for train, info in details if train.dep_name == dep and train.arr_name == arr]
|
||||
filters = [lambda train: train.has_seat()]
|
||||
if include_no_seats:
|
||||
filters.append(lambda train: not train.has_seat())
|
||||
if include_waiting_list:
|
||||
filters.append(lambda train: train.has_waiting_list())
|
||||
trains = [train for train in trains if any(check(train) for check in filters)]
|
||||
if not trains:
|
||||
details = [(train, info) for train, info in details if any(check(train) for check in filters)]
|
||||
if not details:
|
||||
raise NoResultsError()
|
||||
return trains
|
||||
return details
|
||||
|
||||
def search_train(
|
||||
self,
|
||||
dep: str,
|
||||
arr: str,
|
||||
date: str | None = None,
|
||||
time_value: str | None = None,
|
||||
train_type: str = TrainType.ALL,
|
||||
passengers: list[Passenger] | None = None,
|
||||
include_no_seats: bool = False,
|
||||
include_waiting_list: bool = False,
|
||||
):
|
||||
return [
|
||||
train
|
||||
for train, _ in self.search_train_details(
|
||||
dep,
|
||||
arr,
|
||||
date,
|
||||
time_value,
|
||||
train_type=train_type,
|
||||
passengers=passengers,
|
||||
include_no_seats=include_no_seats,
|
||||
include_waiting_list=include_waiting_list,
|
||||
)
|
||||
]
|
||||
|
||||
def train_cars(self, raw_train: dict[str, object], passenger_count: int = 1, room_class: str = "1") -> list[dict[str, object]]:
|
||||
payload = self._seat_lookup_payload(raw_train, passenger_count, room_class)
|
||||
headers, sid = self._auth_headers_and_sid(KORAIL_CARS_INFO)
|
||||
if sid:
|
||||
payload["Sid"] = sid
|
||||
response = self._session.post(KORAIL_CARS_INFO, data=payload, headers=headers)
|
||||
data = json.loads(response.text)
|
||||
if self._result_check(data):
|
||||
cars = data.get("srcar_infos", {}).get("srcar_info", [])
|
||||
if isinstance(cars, dict):
|
||||
cars = [cars]
|
||||
return cars
|
||||
return []
|
||||
|
||||
def car_seats(
|
||||
self,
|
||||
raw_train: dict[str, object],
|
||||
car_no: str,
|
||||
passenger_count: int = 1,
|
||||
room_class: str = "1",
|
||||
) -> dict[str, object]:
|
||||
payload = self._seat_lookup_payload(raw_train, passenger_count, room_class)
|
||||
payload["txtSrcarNo"] = car_no
|
||||
headers, sid = self._auth_headers_and_sid(KORAIL_CAR_DETAIL)
|
||||
if sid:
|
||||
payload["Sid"] = sid
|
||||
response = self._session.post(KORAIL_CAR_DETAIL, data=payload, headers=headers)
|
||||
data = json.loads(response.text)
|
||||
if self._result_check(data):
|
||||
return data
|
||||
return {}
|
||||
|
||||
def _seat_lookup_payload(self, raw_train: dict[str, object], passenger_count: int, room_class: str) -> dict[str, object]:
|
||||
return {
|
||||
"Device": self._device,
|
||||
"Version": self._version,
|
||||
"Key": self._key,
|
||||
"txtArvRsStnCd": raw_train.get("h_arv_rs_stn_cd", ""),
|
||||
"txtArvStnRunOrdr": raw_train.get("h_arv_stn_run_ordr", ""),
|
||||
"txtDptDt": raw_train.get("h_dpt_dt", ""),
|
||||
"txtDptRsStnCd": raw_train.get("h_dpt_rs_stn_cd", ""),
|
||||
"txtDptStnRunOrdr": raw_train.get("h_dpt_stn_run_ordr", ""),
|
||||
"txtGdNo": "",
|
||||
"txtMenuId": "11",
|
||||
"txtPsrmClCd": room_class,
|
||||
"txtRunDt": raw_train.get("h_run_dt", ""),
|
||||
"txtSeatAttCd": "015",
|
||||
"txtTotPsgCnt": str(passenger_count),
|
||||
"txtTrnClsfCd": raw_train.get("h_trn_clsf_cd", ""),
|
||||
"txtTrnGpCd": raw_train.get("h_trn_gp_cd", ""),
|
||||
"txtTrnNo": raw_train.get("h_trn_no", ""),
|
||||
}
|
||||
|
||||
def reserve(self, train, passengers=None, option=ReserveOption.GENERAL_FIRST, try_waiting=False):
|
||||
reserving_seat = True
|
||||
|
|
@ -609,6 +731,14 @@ def find_train_by_id(trains, train_id: str):
|
|||
return None
|
||||
|
||||
|
||||
def find_train_detail_by_id(details, train_id: str):
|
||||
expected = parse_train_id(train_id)
|
||||
for train, raw_train in details:
|
||||
if build_train_id_payload(train) == expected:
|
||||
return train, raw_train
|
||||
return None
|
||||
|
||||
|
||||
def normalize_train(train, index: int) -> dict[str, object]:
|
||||
return {
|
||||
"index": index,
|
||||
|
|
@ -628,6 +758,88 @@ def normalize_train(train, index: int) -> dict[str, object]:
|
|||
}
|
||||
|
||||
|
||||
def parse_seat_label(seat_label: str) -> tuple[int | None, str]:
|
||||
match = re.match(r"^(\d+)([A-Za-z])$", seat_label or "")
|
||||
if not match:
|
||||
return None, ""
|
||||
return int(match.group(1)), match.group(2).upper()
|
||||
|
||||
|
||||
def power_outlet_match(seat_label: str) -> str:
|
||||
row, column = parse_seat_label(seat_label)
|
||||
if row not in POWER_OUTLET_ROWS:
|
||||
return "none"
|
||||
if column in POWER_OUTLET_DIRECT_COLUMNS:
|
||||
return "direct"
|
||||
if column in POWER_OUTLET_ADJACENT_COLUMNS:
|
||||
return "adjacent"
|
||||
return "none"
|
||||
|
||||
|
||||
def normalize_seat(raw_seat: dict[str, object]) -> dict[str, object]:
|
||||
seat_label = str(raw_seat.get("h_con_seat_no", ""))
|
||||
return {
|
||||
"seat": seat_label,
|
||||
"seat_no": str(raw_seat.get("h_seat_no", "")),
|
||||
"available": raw_seat.get("h_sale_psb_flg") == "Y",
|
||||
"direction": SEAT_DIRECTION_NAME.get(str(raw_seat.get("h_for_rev_dir_dv", "")), str(raw_seat.get("h_for_rev_dir_dv", ""))),
|
||||
"position": SEAT_POSITION_NAME.get(str(raw_seat.get("h_sigl_win_in_dv", "")), str(raw_seat.get("h_sigl_win_in_dv", ""))),
|
||||
"seat_type": SEAT_TYPE_NAME.get(str(raw_seat.get("h_dmd_seat_att", "")), str(raw_seat.get("h_dmd_seat_att", ""))),
|
||||
"near_door": raw_seat.get("h_door_nbor_flg") == "Y",
|
||||
"power_outlet": power_outlet_match(seat_label),
|
||||
}
|
||||
|
||||
|
||||
def validate_raw_seat(raw_seat: dict[str, object]) -> None:
|
||||
required_fields = ("h_con_seat_no", "h_seat_no", "h_sale_psb_flg")
|
||||
if any(raw_seat.get(field) in (None, "") for field in required_fields):
|
||||
raise ValueError("seat row is missing required fields")
|
||||
|
||||
|
||||
def parse_nonnegative_int_field(raw: object, field_name: str) -> int:
|
||||
text = "" if raw is None else str(raw)
|
||||
if not text.isdigit():
|
||||
raise ValueError(f"{field_name} is not a non-negative integer")
|
||||
return int(text)
|
||||
|
||||
|
||||
def normalize_car(raw_car: object) -> dict[str, object]:
|
||||
if not isinstance(raw_car, dict):
|
||||
raise ValueError("car row is not an object")
|
||||
return {
|
||||
"car_no": parse_nonnegative_int_field(raw_car.get("h_srcar_no"), "h_srcar_no"),
|
||||
"car_no_raw": str(raw_car.get("h_srcar_no", "")),
|
||||
"room_class": ROOM_CLASS_NAME.get(str(raw_car.get("h_psrm_cl_cd", "")), str(raw_car.get("h_psrm_cl_nm", ""))),
|
||||
"room_class_code": str(raw_car.get("h_psrm_cl_cd", "")),
|
||||
"total_seats": parse_nonnegative_int_field(raw_car.get("h_seat_cnt"), "h_seat_cnt"),
|
||||
"remaining_seats": parse_nonnegative_int_field(raw_car.get("h_rest_seat_cnt"), "h_rest_seat_cnt"),
|
||||
}
|
||||
|
||||
|
||||
def car_center_priority(car: dict[str, object], car_numbers: list[int]) -> tuple[float, int]:
|
||||
car_no = int(car["car_no"])
|
||||
if not car_numbers:
|
||||
return (0.0, car_no)
|
||||
center = (min(car_numbers) + max(car_numbers)) / 2
|
||||
return (abs(car_no - center), car_no)
|
||||
|
||||
|
||||
def sort_cars_for_booking(cars: list[dict[str, object]]) -> list[dict[str, object]]:
|
||||
car_numbers = [int(car["car_no"]) for car in cars]
|
||||
return sorted(cars, key=lambda car: car_center_priority(car, car_numbers))
|
||||
|
||||
|
||||
def seat_preference_key(seat: dict[str, object]) -> tuple[int, int, int, str]:
|
||||
power_rank = {"direct": 0, "adjacent": 1, "none": 2}.get(str(seat.get("power_outlet")), 2)
|
||||
direction_rank = 0 if seat.get("direction") == "순방향" else 1
|
||||
row, column = parse_seat_label(str(seat.get("seat", "")))
|
||||
return (power_rank, direction_rank, row if row is not None else 999, column)
|
||||
|
||||
|
||||
def sort_seats_for_booking(seats: list[dict[str, object]]) -> list[dict[str, object]]:
|
||||
return sorted(seats, key=seat_preference_key)
|
||||
|
||||
|
||||
def mask_identifier(value: object, visible: int = 4) -> str:
|
||||
text = str(value or "")
|
||||
if not text:
|
||||
|
|
@ -699,6 +911,96 @@ def command_search(args: argparse.Namespace) -> None:
|
|||
})
|
||||
|
||||
|
||||
def command_seats(args: argparse.Namespace) -> None:
|
||||
client = build_client()
|
||||
passengers = parse_passengers(args)
|
||||
passenger_count = sum(passenger.count for passenger in Passenger.reduce(passengers))
|
||||
details = client.search_train_details(
|
||||
args.dep,
|
||||
args.arr,
|
||||
args.date,
|
||||
args.time,
|
||||
train_type=TRAIN_TYPE_MAP[args.train_type],
|
||||
passengers=passengers,
|
||||
include_no_seats=True,
|
||||
include_waiting_list=True,
|
||||
)
|
||||
match = find_train_detail_by_id(details, args.train_id)
|
||||
if match is None:
|
||||
raise SystemExit(TRAIN_ID_STALE_MESSAGE)
|
||||
|
||||
train, raw_train = match
|
||||
room_class = ROOM_CLASS_MAP[args.room]
|
||||
seat_car_unavailable = f"seat car data is unavailable for {args.room}; retry search or choose another train"
|
||||
try:
|
||||
cars = [normalize_car(car) for car in client.train_cars(raw_train, passenger_count, room_class)]
|
||||
except (TypeError, ValueError, AttributeError) as exc:
|
||||
raise SystemExit(seat_car_unavailable) from exc
|
||||
if not cars:
|
||||
raise SystemExit(seat_car_unavailable)
|
||||
if args.car_no is not None:
|
||||
cars = [car for car in cars if car["car_no"] == args.car_no]
|
||||
if not cars:
|
||||
raise SystemExit(f"car_no {args.car_no} is not available for {args.room}")
|
||||
else:
|
||||
cars = sort_cars_for_booking(cars)
|
||||
|
||||
car_payloads: list[dict[str, object]] = []
|
||||
for car in cars:
|
||||
raw = client.car_seats(raw_train, str(car["car_no_raw"]), passenger_count, room_class)
|
||||
seat_infos = raw.get("seat_infos") if isinstance(raw, dict) else None
|
||||
seat_detail_unavailable = (
|
||||
f"seat detail data is unavailable for car_no {car['car_no']}; "
|
||||
"retry search or choose another train"
|
||||
)
|
||||
if not isinstance(seat_infos, dict):
|
||||
raise SystemExit(seat_detail_unavailable)
|
||||
if "seat_info" not in seat_infos:
|
||||
raise SystemExit(seat_detail_unavailable)
|
||||
raw_seats = seat_infos["seat_info"]
|
||||
if isinstance(raw_seats, dict):
|
||||
raw_seats = [raw_seats]
|
||||
if not isinstance(raw_seats, list):
|
||||
raise SystemExit(seat_detail_unavailable)
|
||||
if any(not isinstance(seat, dict) for seat in raw_seats):
|
||||
raise SystemExit(seat_detail_unavailable)
|
||||
try:
|
||||
for raw_seat in raw_seats:
|
||||
validate_raw_seat(raw_seat)
|
||||
except ValueError as exc:
|
||||
raise SystemExit(seat_detail_unavailable) from exc
|
||||
remaining_seats = car["remaining_seats"]
|
||||
if not isinstance(remaining_seats, int):
|
||||
raise SystemExit(seat_detail_unavailable)
|
||||
if not raw_seats and remaining_seats > 0:
|
||||
raise SystemExit(seat_detail_unavailable)
|
||||
all_seats = [normalize_seat(seat) for seat in raw_seats if seat.get("h_con_seat_no") != "0A"]
|
||||
if not all_seats and remaining_seats > 0:
|
||||
raise SystemExit(seat_detail_unavailable)
|
||||
seats = sort_seats_for_booking(all_seats)
|
||||
if args.available_only:
|
||||
seats = [seat for seat in seats if seat["available"]]
|
||||
if args.power_only:
|
||||
seats = [seat for seat in seats if seat["power_outlet"] != "none"]
|
||||
available_seats = [seat for seat in seats if seat["available"]]
|
||||
seats = seats[: args.limit]
|
||||
car_payload = dict(car)
|
||||
car_payload["available_seat_count"] = len(available_seats)
|
||||
car_payload["available_seats"] = [seat["seat"] for seat in available_seats]
|
||||
car_payload["shown_seat_count"] = len(seats)
|
||||
car_payload["seats"] = seats
|
||||
car_payloads.append(car_payload)
|
||||
|
||||
print_json({
|
||||
"train": normalize_train(train, 1),
|
||||
"room": args.room,
|
||||
"passenger_count": passenger_count,
|
||||
"available_only": args.available_only,
|
||||
"power_only": args.power_only,
|
||||
"cars": car_payloads,
|
||||
})
|
||||
|
||||
|
||||
def ensure_ncard_available() -> None:
|
||||
if not _NCARD_AVAILABLE:
|
||||
raise SystemExit(
|
||||
|
|
@ -863,6 +1165,33 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
search_parser.add_argument("--include-waiting-list", action="store_true", help="예약 대기 가능 열차도 포함")
|
||||
search_parser.set_defaults(func=command_search)
|
||||
|
||||
seats_parser = subparsers.add_parser("seats", help="조회 결과 중 하나의 호차별 좌석번호를 조회합니다")
|
||||
add_common_trip_args(seats_parser)
|
||||
seats_parser.add_argument("--train-id", required=True, help="search 결과에서 복사한 stable train_id")
|
||||
seats_parser.add_argument(
|
||||
"--room",
|
||||
choices=sorted(ROOM_CLASS_MAP),
|
||||
default="general",
|
||||
help="좌석을 조회할 객실 등급 (기본 general)",
|
||||
)
|
||||
seats_parser.add_argument(
|
||||
"--train-type",
|
||||
choices=sorted(TRAIN_TYPE_MAP),
|
||||
default="ktx",
|
||||
help="재조회할 열차 종류 — search 단계에서 사용한 값과 동일하게 지정 (기본 ktx)",
|
||||
)
|
||||
seats_parser.add_argument("--car-no", type=int, default=None, help="특정 호차만 조회")
|
||||
seats_parser.add_argument(
|
||||
"--available-only",
|
||||
"--remaining-only",
|
||||
dest="available_only",
|
||||
action="store_true",
|
||||
help="예약 가능한/남은 좌석만 출력",
|
||||
)
|
||||
seats_parser.add_argument("--power-only", action="store_true", help="콘센트 꿀팁 좌석(direct/adjacent)만 출력")
|
||||
seats_parser.add_argument("--limit", type=int, default=100, help="호차별 출력할 최대 좌석 수")
|
||||
seats_parser.set_defaults(func=command_seats)
|
||||
|
||||
reserve_parser = subparsers.add_parser("reserve", help="조회 결과 중 하나를 예약합니다")
|
||||
add_common_trip_args(reserve_parser)
|
||||
reserve_parser.add_argument("--train-id", required=True, help="search 결과에서 복사한 stable train_id")
|
||||
|
|
|
|||
|
|
@ -100,12 +100,29 @@ class FakeNCard:
|
|||
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, trains, search_handler=None, ncards=None, ncard_trains=None):
|
||||
def __init__(
|
||||
self,
|
||||
trains,
|
||||
search_handler=None,
|
||||
ncards=None,
|
||||
ncard_trains=None,
|
||||
train_details=None,
|
||||
cars=None,
|
||||
seats_by_car=None,
|
||||
seat_payloads_by_car=None,
|
||||
):
|
||||
self._trains = trains
|
||||
self._search_handler = search_handler
|
||||
self._ncards = ncards or []
|
||||
self._ncard_trains = ncard_trains or []
|
||||
self._train_details = train_details
|
||||
self._cars = cars or []
|
||||
self._seats_by_car = seats_by_car or {}
|
||||
self._seat_payloads_by_car = seat_payloads_by_car or {}
|
||||
self.search_calls = []
|
||||
self.search_detail_calls = []
|
||||
self.train_car_calls = []
|
||||
self.car_seat_calls = []
|
||||
self.reserved_train = None
|
||||
self.reserved_passengers = None
|
||||
|
||||
|
|
@ -115,6 +132,31 @@ class FakeClient:
|
|||
return list(self._search_handler(*args, **kwargs))
|
||||
return list(self._trains)
|
||||
|
||||
def search_train_details(self, *args, **kwargs):
|
||||
self.search_detail_calls.append(kwargs)
|
||||
if self._train_details is not None:
|
||||
return list(self._train_details)
|
||||
return [(train, {}) for train in self._trains]
|
||||
|
||||
def train_cars(self, raw_train, passenger_count=1, room_class="1"):
|
||||
self.train_car_calls.append({
|
||||
"raw_train": raw_train,
|
||||
"passenger_count": passenger_count,
|
||||
"room_class": room_class,
|
||||
})
|
||||
return list(self._cars)
|
||||
|
||||
def car_seats(self, raw_train, car_no, passenger_count=1, room_class="1"):
|
||||
self.car_seat_calls.append({
|
||||
"raw_train": raw_train,
|
||||
"car_no": car_no,
|
||||
"passenger_count": passenger_count,
|
||||
"room_class": room_class,
|
||||
})
|
||||
if car_no in self._seat_payloads_by_car:
|
||||
return self._seat_payloads_by_car[car_no]
|
||||
return {"seat_infos": {"seat_info": list(self._seats_by_car.get(car_no, []))}}
|
||||
|
||||
def reserve(self, train, **kwargs):
|
||||
self.reserved_train = train
|
||||
self.reserved_passengers = kwargs.get("passengers")
|
||||
|
|
@ -173,6 +215,32 @@ class KtxBookingTests(unittest.TestCase):
|
|||
self.assertEqual(args.train_id, "ktx:v1:test")
|
||||
self.assertEqual(args.train_type, "ktx")
|
||||
|
||||
def test_build_parser_accepts_seats_filters(self):
|
||||
args = ktx_booking.build_parser().parse_args([
|
||||
"seats",
|
||||
"서울",
|
||||
"부산",
|
||||
"20260328",
|
||||
"090000",
|
||||
"--train-id",
|
||||
"ktx:v1:test",
|
||||
"--room",
|
||||
"special",
|
||||
"--car-no",
|
||||
"5",
|
||||
"--available-only",
|
||||
"--power-only",
|
||||
"--limit",
|
||||
"10",
|
||||
])
|
||||
|
||||
self.assertEqual(args.train_id, "ktx:v1:test")
|
||||
self.assertEqual(args.room, "special")
|
||||
self.assertEqual(args.car_no, 5)
|
||||
self.assertTrue(args.available_only)
|
||||
self.assertTrue(args.power_only)
|
||||
self.assertEqual(args.limit, 10)
|
||||
|
||||
def test_build_parser_defaults_search_train_type_to_ktx(self):
|
||||
args = ktx_booking.build_parser().parse_args([
|
||||
"search",
|
||||
|
|
@ -207,8 +275,89 @@ class KtxBookingTests(unittest.TestCase):
|
|||
"--train-type",
|
||||
train_type,
|
||||
])
|
||||
seats_args = parser.parse_args([
|
||||
"seats",
|
||||
"서울",
|
||||
"부산",
|
||||
"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)
|
||||
self.assertEqual(seats_args.train_type, train_type)
|
||||
|
||||
def test_normalize_car_and_seat_maps_korail_codes(self):
|
||||
car = ktx_booking.normalize_car({
|
||||
"h_srcar_no": "05",
|
||||
"h_psrm_cl_cd": "1",
|
||||
"h_psrm_cl_nm": "ignored",
|
||||
"h_seat_cnt": "48",
|
||||
"h_rest_seat_cnt": "7",
|
||||
})
|
||||
seat = ktx_booking.normalize_seat({
|
||||
"h_con_seat_no": "7A",
|
||||
"h_seat_no": "007001",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "009",
|
||||
"h_sigl_win_in_dv": "012",
|
||||
"h_dmd_seat_att": "015",
|
||||
"h_door_nbor_flg": "Y",
|
||||
})
|
||||
|
||||
self.assertEqual(car["car_no"], 5)
|
||||
self.assertEqual(car["room_class"], "일반실")
|
||||
self.assertEqual(car["remaining_seats"], 7)
|
||||
self.assertEqual(seat["seat"], "7A")
|
||||
self.assertTrue(seat["available"])
|
||||
self.assertEqual(seat["direction"], "순방향")
|
||||
self.assertEqual(seat["position"], "창측")
|
||||
self.assertEqual(seat["seat_type"], "일반석")
|
||||
self.assertTrue(seat["near_door"])
|
||||
self.assertEqual(seat["power_outlet"], "direct")
|
||||
|
||||
def test_power_outlet_match_distinguishes_direct_adjacent_and_none(self):
|
||||
self.assertEqual(ktx_booking.power_outlet_match("1A"), "direct")
|
||||
self.assertEqual(ktx_booking.power_outlet_match("1B"), "adjacent")
|
||||
self.assertEqual(ktx_booking.power_outlet_match("2A"), "none")
|
||||
self.assertEqual(ktx_booking.power_outlet_match("bad"), "none")
|
||||
|
||||
def test_booking_priority_sorts_middle_cars_before_end_cars(self):
|
||||
cars = [
|
||||
{"car_no": 1},
|
||||
{"car_no": 8},
|
||||
{"car_no": 2},
|
||||
{"car_no": 7},
|
||||
{"car_no": 3},
|
||||
{"car_no": 6},
|
||||
{"car_no": 4},
|
||||
{"car_no": 5},
|
||||
]
|
||||
|
||||
sorted_cars = ktx_booking.sort_cars_for_booking(cars)
|
||||
|
||||
self.assertEqual([car["car_no"] for car in sorted_cars], [4, 5, 3, 6, 2, 7, 1, 8])
|
||||
|
||||
def test_booking_priority_sorts_power_outlet_before_forward_direction(self):
|
||||
seats = [
|
||||
{"seat": "2A", "power_outlet": "none", "direction": "순방향"},
|
||||
{"seat": "1C", "power_outlet": "adjacent", "direction": "역방향"},
|
||||
{"seat": "1A", "power_outlet": "direct", "direction": "역방향"},
|
||||
{"seat": "3A", "power_outlet": "direct", "direction": "순방향"},
|
||||
]
|
||||
|
||||
sorted_seats = ktx_booking.sort_seats_for_booking(seats)
|
||||
|
||||
self.assertEqual([seat["seat"] for seat in sorted_seats], ["3A", "1A", "1C", "2A"])
|
||||
|
||||
def test_is_phone_login_id_accepts_digits_only_mobile_numbers(self):
|
||||
self.assertTrue(ktx_booking.is_phone_login_id("01012345678"))
|
||||
self.assertTrue(ktx_booking.is_phone_login_id("0101234567"))
|
||||
self.assertFalse(ktx_booking.is_phone_login_id("1234567890"))
|
||||
self.assertFalse(ktx_booking.is_phone_login_id("user@example.com"))
|
||||
|
||||
def test_command_search_replays_selected_train_type(self):
|
||||
selected = FakeTrain(
|
||||
|
|
@ -316,6 +465,713 @@ class KtxBookingTests(unittest.TestCase):
|
|||
self.assertTrue(client.search_calls[-1]["include_waiting_list"])
|
||||
self.assertIs(client.reserved_train, waiting_only)
|
||||
|
||||
def test_command_seats_returns_available_power_seats_for_selected_car(self):
|
||||
selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")
|
||||
raw_train = {"h_trn_no": "009", "h_dpt_dt": "20260328"}
|
||||
train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(selected, raw_train)],
|
||||
cars=[
|
||||
{
|
||||
"h_srcar_no": "04",
|
||||
"h_psrm_cl_cd": "1",
|
||||
"h_seat_cnt": "48",
|
||||
"h_rest_seat_cnt": "9",
|
||||
},
|
||||
{
|
||||
"h_srcar_no": "05",
|
||||
"h_psrm_cl_cd": "1",
|
||||
"h_seat_cnt": "48",
|
||||
"h_rest_seat_cnt": "3",
|
||||
},
|
||||
],
|
||||
seats_by_car={
|
||||
"05": [
|
||||
{
|
||||
"h_con_seat_no": "1A",
|
||||
"h_seat_no": "001001",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "009",
|
||||
"h_sigl_win_in_dv": "012",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
{
|
||||
"h_con_seat_no": "1B",
|
||||
"h_seat_no": "001002",
|
||||
"h_sale_psb_flg": "N",
|
||||
"h_for_rev_dir_dv": "010",
|
||||
"h_sigl_win_in_dv": "013",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
{
|
||||
"h_con_seat_no": "2A",
|
||||
"h_seat_no": "002001",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "009",
|
||||
"h_sigl_win_in_dv": "012",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
{
|
||||
"h_con_seat_no": "0A",
|
||||
"h_seat_no": "000000",
|
||||
"h_sale_psb_flg": "Y",
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=2,
|
||||
children=1,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=5,
|
||||
available_only=True,
|
||||
power_only=True,
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
self.assertEqual(result["room"], "general")
|
||||
self.assertEqual(result["passenger_count"], 3)
|
||||
self.assertTrue(result["available_only"])
|
||||
self.assertTrue(result["power_only"])
|
||||
self.assertEqual(len(result["cars"]), 1)
|
||||
self.assertEqual(result["cars"][0]["car_no"], 5)
|
||||
self.assertEqual(result["cars"][0]["remaining_seats"], 3)
|
||||
self.assertEqual(result["cars"][0]["available_seat_count"], 1)
|
||||
self.assertEqual(result["cars"][0]["available_seats"], ["1A"])
|
||||
self.assertEqual(result["cars"][0]["shown_seat_count"], 1)
|
||||
self.assertEqual(result["cars"][0]["seats"][0]["seat"], "1A")
|
||||
self.assertEqual(result["cars"][0]["seats"][0]["power_outlet"], "direct")
|
||||
self.assertEqual(client.search_detail_calls[-1]["train_type"], ktx_booking.TRAIN_TYPE_MAP["ktx"])
|
||||
self.assertTrue(client.search_detail_calls[-1]["include_no_seats"])
|
||||
self.assertTrue(client.search_detail_calls[-1]["include_waiting_list"])
|
||||
self.assertEqual(client.train_car_calls[-1]["passenger_count"], 3)
|
||||
self.assertEqual(client.train_car_calls[-1]["room_class"], "1")
|
||||
self.assertEqual(client.car_seat_calls[-1]["car_no"], "05")
|
||||
|
||||
def test_command_seats_explores_middle_cars_first(self):
|
||||
selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")
|
||||
raw_train = {"h_trn_no": "009", "h_dpt_dt": "20260328"}
|
||||
train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(selected, raw_train)],
|
||||
cars=[
|
||||
{"h_srcar_no": "01", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "1"},
|
||||
{"h_srcar_no": "08", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "1"},
|
||||
{"h_srcar_no": "04", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "1"},
|
||||
{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "1"},
|
||||
],
|
||||
seats_by_car={
|
||||
car_no: [{"h_con_seat_no": "1A", "h_seat_no": "001001", "h_sale_psb_flg": "Y"}]
|
||||
for car_no in ("01", "04", "05", "08")
|
||||
},
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
self.assertEqual([car["car_no"] for car in result["cars"]], [4, 5, 1, 8])
|
||||
self.assertEqual([call["car_no"] for call in client.car_seat_calls], ["04", "05", "01", "08"])
|
||||
|
||||
def test_command_seats_outputs_available_seats_by_booking_preference(self):
|
||||
selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")
|
||||
raw_train = {"h_trn_no": "009", "h_dpt_dt": "20260328"}
|
||||
train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(selected, raw_train)],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "4"}],
|
||||
seats_by_car={
|
||||
"05": [
|
||||
{
|
||||
"h_con_seat_no": "2A",
|
||||
"h_seat_no": "002001",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "009",
|
||||
"h_sigl_win_in_dv": "012",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
{
|
||||
"h_con_seat_no": "1C",
|
||||
"h_seat_no": "001003",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "010",
|
||||
"h_sigl_win_in_dv": "013",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
{
|
||||
"h_con_seat_no": "1A",
|
||||
"h_seat_no": "001001",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "010",
|
||||
"h_sigl_win_in_dv": "012",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
{
|
||||
"h_con_seat_no": "3A",
|
||||
"h_seat_no": "003001",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "009",
|
||||
"h_sigl_win_in_dv": "012",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=True,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
car = result["cars"][0]
|
||||
self.assertEqual(car["available_seats"], ["3A", "1A", "1C", "2A"])
|
||||
self.assertEqual([seat["seat"] for seat in car["seats"]], ["3A", "1A", "1C", "2A"])
|
||||
|
||||
def test_command_seats_available_summary_matches_power_filter(self):
|
||||
selected = FakeTrain(train_no="009", dep_time="090000", arr_time="113000", label="selected")
|
||||
raw_train = {"h_trn_no": "009", "h_dpt_dt": "20260328"}
|
||||
train_id = ktx_booking.normalize_train(selected, index=1)["train_id"]
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(selected, raw_train)],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "4"}],
|
||||
seats_by_car={
|
||||
"05": [
|
||||
{
|
||||
"h_con_seat_no": "1A",
|
||||
"h_seat_no": "001001",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "009",
|
||||
"h_sigl_win_in_dv": "012",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
{
|
||||
"h_con_seat_no": "2C",
|
||||
"h_seat_no": "002003",
|
||||
"h_sale_psb_flg": "Y",
|
||||
"h_for_rev_dir_dv": "009",
|
||||
"h_sigl_win_in_dv": "013",
|
||||
"h_dmd_seat_att": "015",
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=True,
|
||||
power_only=True,
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
car = json.loads(output.getvalue())["cars"][0]
|
||||
self.assertEqual(car["available_seat_count"], 1)
|
||||
self.assertEqual(car["available_seats"], ["1A"])
|
||||
self.assertEqual([seat["seat"] for seat in car["seats"]], ["1A"])
|
||||
|
||||
def test_command_seats_supports_special_room_and_stale_train_error(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"]
|
||||
other = FakeTrain(train_no="011", dep_time="093000", arr_time="120000", label="other")
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(other, {"h_trn_no": "011"})],
|
||||
cars=[{"h_srcar_no": "01", "h_psrm_cl_cd": "2", "h_seat_cnt": "30", "h_rest_seat_cnt": "1"}],
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="special",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("train_id", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_requested_car_is_not_available(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(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": "04", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"}],
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=5,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("car_no 5", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_seat_payload_is_malformed(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(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "4"}],
|
||||
)
|
||||
client.car_seats = lambda *args, **kwargs: {"seat_infos": None}
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat detail data is unavailable", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_seat_info_key_is_missing(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(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"}],
|
||||
)
|
||||
client.car_seats = lambda *args, **kwargs: {"seat_infos": {}}
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat detail data is unavailable", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_remaining_seats_have_empty_seat_info(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(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"}],
|
||||
seat_payloads_by_car={"05": {"seat_infos": {"seat_info": []}}},
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat detail data is unavailable", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_seat_info_contains_non_object_entries(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"]
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
for bad_entry in ["bad", None]:
|
||||
with self.subTest(bad_entry=bad_entry):
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"}],
|
||||
seat_payloads_by_car={"05": {"seat_infos": {"seat_info": [bad_entry]}}},
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat detail data is unavailable", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_remaining_seats_have_only_sentinel_seat_info(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(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"}],
|
||||
seat_payloads_by_car={
|
||||
"05": {
|
||||
"seat_infos": {
|
||||
"seat_info": [
|
||||
{"h_con_seat_no": "0A", "h_seat_no": "000000", "h_sale_psb_flg": "N"},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat detail data is unavailable", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_seat_info_object_is_missing_required_fields(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"]
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
malformed_rows = [
|
||||
{},
|
||||
{"h_seat_no": "001001", "h_sale_psb_flg": "Y"},
|
||||
{"h_con_seat_no": "1A", "h_sale_psb_flg": "Y"},
|
||||
{"h_con_seat_no": "1A", "h_seat_no": "001001"},
|
||||
]
|
||||
|
||||
for row in malformed_rows:
|
||||
with self.subTest(row=row):
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"}],
|
||||
seat_payloads_by_car={"05": {"seat_infos": {"seat_info": [row]}}},
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat detail data is unavailable", str(exc.exception))
|
||||
|
||||
def test_command_seats_allows_empty_seat_info_when_no_remaining_seats(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(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[{"h_srcar_no": 5, "h_psrm_cl_cd": "1", "h_seat_cnt": 48, "h_rest_seat_cnt": 0}],
|
||||
seat_payloads_by_car={"5": {"seat_infos": {"seat_info": []}}},
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
output = io.StringIO()
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with redirect_stdout(output):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
car = json.loads(output.getvalue())["cars"][0]
|
||||
self.assertEqual(car["remaining_seats"], 0)
|
||||
self.assertEqual(car["available_seat_count"], 0)
|
||||
self.assertEqual(car["seats"], [])
|
||||
|
||||
def test_command_seats_fails_when_car_metadata_is_malformed(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"]
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
malformed_cars = [
|
||||
"bad",
|
||||
{"h_srcar_no": "bad", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "9"},
|
||||
{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "bad", "h_rest_seat_cnt": "9"},
|
||||
{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48", "h_rest_seat_cnt": "bad"},
|
||||
{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_rest_seat_cnt": "9"},
|
||||
{"h_srcar_no": "05", "h_psrm_cl_cd": "1", "h_seat_cnt": "48"},
|
||||
]
|
||||
|
||||
for raw_car in malformed_cars:
|
||||
with self.subTest(raw_car=raw_car):
|
||||
client = FakeClient(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[raw_car],
|
||||
seats_by_car={"05": [{"h_con_seat_no": "1A", "h_seat_no": "001001", "h_sale_psb_flg": "Y"}]},
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat car data is unavailable", str(exc.exception))
|
||||
|
||||
def test_command_seats_fails_when_car_data_is_unavailable(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(
|
||||
[],
|
||||
train_details=[(selected, {"h_trn_no": "009"})],
|
||||
cars=[],
|
||||
)
|
||||
args = argparse.Namespace(
|
||||
dep="서울",
|
||||
arr="부산",
|
||||
date="20260328",
|
||||
time="090000",
|
||||
adults=1,
|
||||
children=0,
|
||||
toddlers=0,
|
||||
seniors=0,
|
||||
train_id=train_id,
|
||||
room="general",
|
||||
train_type="ktx",
|
||||
car_no=None,
|
||||
available_only=False,
|
||||
power_only=False,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
with patch.object(ktx_booking, "build_client", return_value=client):
|
||||
with self.assertRaises(SystemExit) as exc:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
ktx_booking.command_seats(args)
|
||||
|
||||
self.assertIn("seat car data is unavailable", str(exc.exception))
|
||||
|
||||
def test_seat_research_endpoints_use_dynapath_sid_boundary(self):
|
||||
class FakeEngine:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def generate_token(self, device_id, timestamp_ms, nonce):
|
||||
self.calls.append((device_id, timestamp_ms, nonce))
|
||||
return "dynapath-token"
|
||||
|
||||
client = ktx_booking.PatchedKorail.__new__(ktx_booking.PatchedKorail)
|
||||
client._engine = FakeEngine()
|
||||
client._device_id = "device-id"
|
||||
client._generate_sid = lambda timestamp_ms: f"sid-{timestamp_ms}"
|
||||
|
||||
for url in (ktx_booking.KORAIL_CARS_INFO, ktx_booking.KORAIL_CAR_DETAIL):
|
||||
with self.subTest(url=url):
|
||||
with patch.object(ktx_booking.time, "time", return_value=1234.567):
|
||||
with patch.object(ktx_booking.random, "choices", return_value=list("ABCD")):
|
||||
headers, sid = client._auth_headers_and_sid(url)
|
||||
|
||||
self.assertEqual(headers["x-dynapath-m-token"], "dynapath-token")
|
||||
self.assertEqual(sid, "sid-1234567")
|
||||
|
||||
self.assertEqual(
|
||||
client._engine.calls,
|
||||
[("device-id", 1234567, "ABCD"), ("device-id", 1234567, "ABCD")],
|
||||
)
|
||||
|
||||
def test_build_parser_has_ncard_commands(self):
|
||||
parser = ktx_booking.build_parser()
|
||||
help_text = parser.format_help()
|
||||
|
|
|
|||
|
|
@ -36,10 +36,12 @@ while IFS= read -r -d '' skill_dir; do
|
|||
fi
|
||||
done < <(
|
||||
find "$root" -mindepth 1 -maxdepth 1 -type d \
|
||||
! -name '.*' \
|
||||
! -name .git \
|
||||
! -name .github \
|
||||
! -name .codex \
|
||||
! -name .claude \
|
||||
! -name .agents \
|
||||
! -name .omx \
|
||||
! -name .ouroboros \
|
||||
! -name .changeset \
|
||||
|
|
@ -47,6 +49,7 @@ done < <(
|
|||
! -name .vscode \
|
||||
! -name .sisyphus \
|
||||
! -name .idea \
|
||||
! -name .venv \
|
||||
! -name dist \
|
||||
! -name docs \
|
||||
! -name node_modules \
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue