pytest for Map Reduce

conftest.py

@pytest.fixture
def tmp_dir():
    root_path = os.getcwd()
    letters = string.ascii_lowercase
    tmp_name = "".join(choice(letters) for _ in range(8))
    return os.path.join(root_path, tmp_name)
@pytest.fixture
def make_test_files(tmp_dir):
    print(f"\ntest_files at {tmp_dir}")
    os.makedirs(tmp_dir, exist_ok=True)
    letters = string.ascii_lowercase
    total_count = 0
    for i in range(10):
        with open(os.path.join(tmp_dir, "{:02d}.txt".format(i)), "w") as wf:
            for j in range(randint(100, 200)):
                line: str = "".join(choice(letters) for k in range(160)) + "\n"
                total_count += 1
                wf.write(line)
    return total_count
@pytest.fixture
def test_files_ready(request, make_test_files, tmp_dir):
    total_count = make_test_files
    def teardown_remove_temp():
        print(f"\nremove {tmp_dir}")
        remove_dir(tmp_dir)
    request.addfinalizer(teardown_remove_temp)
    return total_count
@pytest.fixture
def elapsed_time():
    start = datetime.now()
    yield
    elapsed = datetime.now() - start
    print(f"\nruntime={elapsed.total_seconds():.3f}sec")

Map Reduce

Input Class 선언

class InputData(object):
    def read(self):
        raise NotImplementedError

    @classmethod
    def generate_inputs(cls, file_path: str):
        raise NotImplementedError


class FileInputData(InputData):
    def __init__(self, file_path):
        if not file_path:
            raise ValueError(f"path error : {file_path}")
        super().__init__()

        if not os.path.isfile(file_path):
            raise ValueError(f"file not exists {file_path}")

        self.path = file_path

    def read(self):
        return open(self.path).read()

    @classmethod
    def generate_inputs(cls, path: str):
        for name in os.listdir(path):
            yield cls(os.path.join(path, name))

List Comprehension

Lucky
한 밤에 중앙 공원에서 …(럭키의 하루!)

map과 filter 대신에 리스트 컴프리헨션을 사용하자

리스트 a에 있는 각 숫자의 제곱을 계산할 경우