byeongwook.choiandClaude Fable 5.1
fdae34161e
- GET_PROGRAM_LIST 는 TADIR 의 PROG 만 주므로 GET_OBJECT_LIST_BY_PACKAGE(FUGR) 로 그룹을 받아 SAPL<그룹> 을 프로그램처럼 저장 (설명에 펑션 이름·텍스트 목록) - 펑션 그룹은 서버 INCLUDE_LIST 가 비어 온다(실측) → 소스의 INCLUDE 문을 깊이 4 까지 따라가며 고객 인클루드(Z/Y/네임스페이스, L<그룹>…)만 받는다. 표준 생성기 인클루드(LSVIM…)는 제외 - 실측 ZFI01: 펑션 그룹 14, SAPLZFG_FI01 인클루드 47 → 파서 FUNCTION 38 unit, 미인식 0 - --no-fugr 로 끌 수 있다 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
231 lines
12 KiB
Python
231 lines
12 KiB
Python
"""Stage 0 (ingest/from_sap.py) — SAP 응답을 가짜로 넣어 저장 형식과 normalize 연결을 확인한다.
|
||
|
||
실제 SAP 은 부르지 않는다. 응답 모양은 ../abap-api-tester/tools/spec/samples 의 실측 JSON 과 같다.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from ingest import from_sap
|
||
from ingest.normalize import classify, parse_collected_file, run as normalize_run
|
||
|
||
PROGRAM_LIST = {
|
||
"RETURN": {"TYPE": "S", "MESSAGE": "", "TOTAL_ROWS": "2"},
|
||
"RESULT": [
|
||
{"OBJ_NAME": "ZFIR0010", "TEXT": "거래처 I/F 이력", "SUBC": "1", "UDAT": "2024-01-02", "UNAM": "U1"},
|
||
{"OBJ_NAME": "SAPMZSFT0", "TEXT": "달력 유지보수", "SUBC": "M", "UDAT": "2020-10-16", "UNAM": "U2"},
|
||
],
|
||
}
|
||
|
||
SOURCES = {
|
||
"ZFIR0010": {
|
||
"RETURN": {"TYPE": "S", "MESSAGE": "", "TOTAL_ROWS": "0"},
|
||
"PROGRAM": "ZFIR0010",
|
||
"SOURCE_CODE": "REPORT zfir0010.\nINCLUDE zfir0010_top.\nSTART-OF-SELECTION.\n PERFORM main.\n",
|
||
"INCLUDE_LIST": [
|
||
{"INCL_NAME": "ZFIR0010_TOP", "SOURCE": "DATA gv_bukrs TYPE bukrs.\n"},
|
||
{"INCL_NAME": "DB__SSEL", "SOURCE": ""}, # 빈 소스는 버린다
|
||
{"INCL_NAME": "ZFIR0010_TOP", "SOURCE": "DATA dup.\n"}, # 중복은 버린다
|
||
],
|
||
"SCREEN_LIST": "",
|
||
"TCODE_LIST": [{"TCODE": "ZFIR0010", "TTEXT": "거래처 I/F 이력조회", "PGMNA": "ZFIR0010"}],
|
||
},
|
||
"SAPMZSFT0": {
|
||
"RETURN": {"TYPE": "E", "MESSAGE": "프로그램 소스를 읽을 수 없습니다", "TOTAL_ROWS": "0"},
|
||
"PROGRAM": "SAPMZSFT0", "SOURCE_CODE": "", "INCLUDE_LIST": "", "SCREEN_LIST": "", "TCODE_LIST": "",
|
||
},
|
||
}
|
||
|
||
|
||
FUGR_LIST = {"RETURN": {"TYPE": "S", "MESSAGE": "", "TOTAL_ROWS": 1},
|
||
"RESULT": [{"OBJECT": "FUGR", "OBJ_NAME": "ZFG_FI01", "AUTHOR": "U1"}]}
|
||
FUNC_LIST = {"RETURN": {"TYPE": "S", "MESSAGE": "", "TOTAL_ROWS": 1},
|
||
"RESULT": [{"FUNCNAME": "Z_FI_CHECK", "STEXT": "사업자번호 체크"}]}
|
||
# 펑션 그룹: 서버 INCLUDE_LIST 는 비어 오고, 소스의 INCLUDE 문을 따라가야 한다 (실측)
|
||
SOURCES["SAPLZFG_FI01"] = {
|
||
"RETURN": {"TYPE": "S", "MESSAGE": "", "TOTAL_ROWS": 0}, "PROGRAM": "SAPLZFG_FI01",
|
||
"SOURCE_CODE": "* regenerated\nINCLUDE LZFG_FI01TOP.\nINCLUDE LZFG_FI01UXX.\nINCLUDE LSVIMFXX.\n* INCLUDE LZFG_FI01ZZZ.\n",
|
||
"INCLUDE_LIST": [], "SCREEN_LIST": [], "TCODE_LIST": [],
|
||
}
|
||
SOURCES["LZFG_FI01TOP"] = {"RETURN": {"TYPE": "S", "MESSAGE": ""}, "PROGRAM": "LZFG_FI01TOP",
|
||
"SOURCE_CODE": "FUNCTION-POOL zfg_fi01.\nINCLUDE lsvimdat.\n", "INCLUDE_LIST": []}
|
||
SOURCES["LZFG_FI01UXX"] = {"RETURN": {"TYPE": "S", "MESSAGE": ""}, "PROGRAM": "LZFG_FI01UXX",
|
||
"SOURCE_CODE": "INCLUDE LZFG_FI01U01.\n", "INCLUDE_LIST": []}
|
||
SOURCES["LZFG_FI01U01"] = {"RETURN": {"TYPE": "S", "MESSAGE": ""}, "PROGRAM": "LZFG_FI01U01",
|
||
"SOURCE_CODE": "FUNCTION z_fi_check.\n SELECT SINGLE * FROM t001.\nENDFUNCTION.\n",
|
||
"INCLUDE_LIST": []}
|
||
CALLED: list[str] = []
|
||
|
||
|
||
def fake_sap_call(method: str, params: dict, timeout: int = 180) -> dict:
|
||
CALLED.append(f"{method}:{params.get('IV_PROGRAM') or params.get('IV_OBJ_TYPE') or params.get('IV_FUNC_GROUP') or ''}")
|
||
if method == "GET_PROGRAM_LIST":
|
||
assert params["IV_PACKAGE"] == "ZFI01"
|
||
data = PROGRAM_LIST
|
||
elif method == "GET_OBJECT_LIST_BY_PACKAGE":
|
||
assert params["IV_OBJ_TYPE"] == "FUGR"
|
||
data = FUGR_LIST
|
||
elif method == "GET_FUNCTION_LIST":
|
||
data = FUNC_LIST
|
||
elif method == "GET_PROGRAM_SOURCE":
|
||
name = params["IV_PROGRAM"]
|
||
if name not in SOURCES: # 표준 인클루드(LSVIMFXX 등)는 요청 자체가 오면 안 된다
|
||
raise AssertionError(f"요청하면 안 되는 소스: {name}")
|
||
data = SOURCES[name]
|
||
else:
|
||
raise AssertionError(method)
|
||
ret = data["RETURN"]
|
||
return {"method": method, "parsed": data, "error": None,
|
||
"return": {"type": ret["TYPE"], "message": ret["MESSAGE"]}}
|
||
|
||
|
||
@pytest.fixture
|
||
def sap(monkeypatch):
|
||
import sap.sap_client as client
|
||
monkeypatch.setattr(client, "sap_call", fake_sap_call)
|
||
monkeypatch.setattr(from_sap.time, "sleep", lambda s: None)
|
||
|
||
|
||
def test_run_writes_raw_files_and_records_errors(tmp_path, sap):
|
||
stats = from_sap.run(["zfi01"], tmp_path, fugr=False)
|
||
assert stats == {"packages": 1, "listed": 2, "function_groups": 0, "fetched": 1, "skipped": 0,
|
||
"errors": 1, "includes": 2}
|
||
|
||
# 패키지 목록: normalize 가 package_list 로 분류하는 모양
|
||
pkg = parse_collected_file((tmp_path / "ZFI01.txt").read_text(encoding="utf-8"))
|
||
assert classify(pkg) == "package_list"
|
||
assert pkg[0] == {"DEVCLASS": "ZFI01", "OBJ_NAME": "ZFIR0010", "TEXT": "거래처 I/F 이력",
|
||
"CREATED_ON": "", "CHANGED_ON": "2024-01-02"}
|
||
|
||
# 프로그램 소스: 메인 + 인클루드(빈 것·중복 제외), 줄바꿈은 JSON 이스케이프로 보존
|
||
prog = parse_collected_file((tmp_path / "ZFIR0010.txt").read_text(encoding="utf-8"))
|
||
assert classify(prog) == "program_source"
|
||
assert prog["DESCRIPTION"] == "거래처 I/F 이력"
|
||
assert [i["INCLUDE"] for i in prog["INCLUDE_PROGRAM"]] == ["ZFIR0010", "ZFIR0010_TOP"]
|
||
assert prog["INCLUDE_PROGRAM"][0]["SOURCE_CODE"].splitlines()[0] == "REPORT zfir0010."
|
||
assert prog["TEXT_SYMBOL"] == []
|
||
assert prog["TCODE_LIST"][0]["TCODE"] == "ZFIR0010"
|
||
|
||
# 실패한 프로그램은 파일 없이 _errors.log 에
|
||
assert not (tmp_path / "SAPMZSFT0.txt").exists()
|
||
assert "SAPMZSFT0\tSOURCE\tSAP 오류" in (tmp_path / "_errors.log").read_text(encoding="utf-8")
|
||
|
||
|
||
def test_rerun_skips_existing_unless_force(tmp_path, sap):
|
||
from_sap.run(["ZFI01"], tmp_path, fugr=False)
|
||
again = from_sap.run(["ZFI01"], tmp_path, fugr=False)
|
||
assert again["skipped"] == 1 and again["fetched"] == 0
|
||
forced = from_sap.run(["ZFI01"], tmp_path, force=True, fugr=False)
|
||
assert forced["fetched"] == 1
|
||
|
||
|
||
def test_normalize_consumes_output_and_builds_tcodes(tmp_path, sap):
|
||
raw, out = tmp_path / "raw", tmp_path / "normalized"
|
||
from_sap.run(["ZFI01"], raw, fugr=False)
|
||
stats = normalize_run(raw, out)
|
||
assert stats["programs"] == 1 and stats["package_rows"] == 2 and stats["errors"] == 0
|
||
|
||
assert (out / "ZFIR0010" / "ZFIR0010.abap").read_text(encoding="utf-8").startswith("REPORT zfir0010.\n")
|
||
assert (out / "ZFIR0010" / "ZFIR0010_TOP.abap").exists()
|
||
meta = json.loads((out / "ZFIR0010" / "ZFIR0010.meta.json").read_text(encoding="utf-8"))
|
||
assert meta["description"] == "거래처 I/F 이력" and len(meta["includes"]) == 2
|
||
|
||
tcodes = [json.loads(l) for l in (out / "tcodes.jsonl").read_text(encoding="utf-8").splitlines()]
|
||
assert tcodes == [{"tcode": "ZFIR0010", "program": "ZFIR0010", "text_ko": "거래처 I/F 이력조회"}]
|
||
|
||
|
||
def test_dry_run_writes_nothing(tmp_path, sap):
|
||
stats = from_sap.run(["ZFI01"], tmp_path, dry_run=True, fugr=False)
|
||
assert stats["listed"] == 2 and stats["fetched"] == 0
|
||
assert list(tmp_path.iterdir()) == []
|
||
|
||
|
||
def test_sap_client_parses_json_and_xml_responses():
|
||
from sap.sap_client import parse_response
|
||
|
||
js = parse_response('{"RETURN":{"TYPE":"W","MESSAGE":"잘림","TOTAL_ROWS":177},"RESULT":[{"OBJ_NAME":"ZFI0000"}],"INCLUDE_LIST":[]}'.encode("utf-8"))
|
||
assert js["error"] is None and js["return"] == {"type": "W", "message": "잘림"}
|
||
assert js["parsed"]["RESULT"][0]["OBJ_NAME"] == "ZFI0000" and js["parsed"]["INCLUDE_LIST"] == []
|
||
|
||
xml = parse_response(b'\xef\xbb\xbf<?xml version="1.0" encoding="utf-16"?><asx:abap xmlns:asx="x"><asx:values><DATA>'
|
||
b'<RETURN><TYPE>S</TYPE><MESSAGE/></RETURN><RESULT/></DATA></asx:values></asx:abap>')
|
||
assert xml["error"] is None and xml["return"]["type"] == "S" and xml["parsed"]["RESULT"] == []
|
||
|
||
bad = parse_response(b"<html>login</html>")
|
||
assert bad["return"]["type"] == "X"
|
||
|
||
|
||
def test_sap_call_sends_fields_without_wrapper(monkeypatch):
|
||
import sap.sap_client as client
|
||
|
||
captured = {}
|
||
|
||
class _Resp:
|
||
status = 200
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): return False
|
||
def read(self): return b'{"RETURN":{"TYPE":"S","MESSAGE":"","TOTAL_ROWS":0},"RESULT":[]}'
|
||
|
||
def fake_urlopen(req, context=None, timeout=None):
|
||
captured["body"] = json.loads(req.data.decode("utf-8"))
|
||
captured["url"] = req.full_url
|
||
return _Resp()
|
||
|
||
monkeypatch.setattr(client.urllib.request, "urlopen", fake_urlopen)
|
||
res = client.sap_call("GET_PROGRAM_LIST", {"IV_PACKAGE": "ZFI01", "IV_PATTERN": "", "IV_MAX_ROWS": 5})
|
||
assert captured["body"] == {"IV_PACKAGE": "ZFI01", "IV_MAX_ROWS": 5} # 래퍼 없음, 빈 값 제외, 숫자 유지
|
||
assert captured["url"].endswith("/GET_PROGRAM_LIST")
|
||
assert res["return"]["type"] == "S" and res["parsed"]["RESULT"] == []
|
||
|
||
|
||
def test_normalize_survives_non_string_fields_and_bad_file(tmp_path):
|
||
"""JSON 규약 응답엔 숫자·null 이 섞인다. 파일 하나가 깨져도 나머지는 계속 처리해야 한다."""
|
||
from ingest.normalize import clean_text_field
|
||
assert clean_text_field(None) == "" and clean_text_field(20240102) == "20240102" and clean_text_field(0) == "0"
|
||
|
||
raw, out = tmp_path / "raw", tmp_path / "normalized"
|
||
raw.mkdir()
|
||
(raw / "ZPKG.txt").write_text(json.dumps([
|
||
{"DEVCLASS": "ZPKG", "OBJ_NAME": "ZA", "TEXT": None, "CREATED_ON": "", "CHANGED_ON": 20240102},
|
||
{"DEVCLASS": "ZPKG", "OBJ_NAME": "ZB", "TEXT": "b", "CREATED_ON": None, "CHANGED_ON": None},
|
||
]), encoding="utf-8")
|
||
(raw / "ZBAD.txt").write_text('{"MAIN_PROGRAM": "ZBAD", "INCLUDE_PROGRAM": "not-a-list"}', encoding="utf-8")
|
||
(raw / "ZOK.txt").write_text(json.dumps({"MAIN_PROGRAM": "ZOK", "DESCRIPTION": "ok",
|
||
"INCLUDE_PROGRAM": [{"INCLUDE": "ZOK", "SOURCE_CODE": "REPORT zok.\n"}]}),
|
||
encoding="utf-8")
|
||
stats = normalize_run(raw, out)
|
||
assert stats["programs"] == 1 and stats["package_rows"] == 2 and stats["errors"] == 1
|
||
assert (out / "ZOK" / "ZOK.abap").exists()
|
||
rows = [json.loads(l) for l in (out / "packages.jsonl").read_text(encoding="utf-8").splitlines()]
|
||
assert rows[0]["changed_on"] == "20240102" and rows[0]["text"] == "" and rows[1]["changed_on"] == ""
|
||
assert "ZBAD.txt" in (out / "_errors.log").read_text(encoding="utf-8")
|
||
|
||
|
||
def test_function_groups_are_fetched_via_include_statements(tmp_path, sap):
|
||
"""펑션 그룹: FUGR 목록 → SAPL<그룹> 저장, 인클루드는 INCLUDE 문을 따라(표준 인클루드는 제외, 주석은 무시)."""
|
||
CALLED.clear()
|
||
stats = from_sap.run(["ZFI01"], tmp_path)
|
||
assert stats["function_groups"] == 1 and stats["listed"] == 3 and stats["fetched"] == 2
|
||
prog = parse_collected_file((tmp_path / "SAPLZFG_FI01.txt").read_text(encoding="utf-8"))
|
||
assert prog["MAIN_PROGRAM"] == "SAPLZFG_FI01"
|
||
assert prog["DESCRIPTION"] == "펑션 그룹 ZFG_FI01: Z_FI_CHECK (사업자번호 체크)"
|
||
assert [i["INCLUDE"] for i in prog["INCLUDE_PROGRAM"]] == [
|
||
"SAPLZFG_FI01", "LZFG_FI01TOP", "LZFG_FI01UXX", "LZFG_FI01U01"]
|
||
assert "ENDFUNCTION" in prog["INCLUDE_PROGRAM"][3]["SOURCE_CODE"]
|
||
assert not any(c.endswith(":LSVIMFXX") or c.endswith(":LSVIMDAT") or c.endswith(":LZFG_FI01ZZZ") for c in CALLED)
|
||
# 패키지 목록에도 SAPL 프로그램이 들어간다 (package 테이블 조인용)
|
||
pkg = parse_collected_file((tmp_path / "ZFI01.txt").read_text(encoding="utf-8"))
|
||
assert any(r["OBJ_NAME"] == "SAPLZFG_FI01" for r in pkg)
|
||
|
||
|
||
def test_is_customer_include():
|
||
from ingest.from_sap import is_customer_include
|
||
assert is_customer_include("LZFG_FI01U01", "SAPLZFG_FI01")
|
||
assert is_customer_include("ZFI_COMMON", "ZFI0000")
|
||
assert is_customer_include("/PWC/ZINC", "ZFI0000")
|
||
assert not is_customer_include("LSVIMFXX", "SAPLZFG_FI01")
|
||
assert not is_customer_include("LZOTHERU01", "SAPLZFG_FI01") # 다른 그룹의 L… 인클루드는 제외
|
||
assert not is_customer_include("RSDBRUNT", "ZFI0000")
|