首先，校验 yaml 文件的格式，返回 yaml 的数据。

def load_yaml_file(yaml_file):
    """ load yaml file and check file content format
    """
    with io.open(yaml_file, 'r', encoding='utf-8') as stream:
        yaml_content = yaml.load(stream)
        _check_format(yaml_file, yaml_content)
        return yaml_content

接下来，根据你的测试指令后的文件的后缀名，读取对应格式的文件。

def load_file(file_path):
    if not os.path.isfile(file_path):
        raise exceptions.FileNotFound("{} does not exist.".format(file_path))

    file_suffix = os.path.splitext(file_path)[1].lower()
    if file_suffix == '.json':
        return load_json_file(file_path)
    elif file_suffix in ['.yaml', '.yml']:
        return load_yaml_file(file_path)
    elif file_suffix == ".csv":
        return load_csv_file(file_path)
    else:
        # '' or other suffix
        err_msg = u"Unsupported file format: {}".format(file_path)
        logger.log_warning(err_msg)
        return []

然后，下面这个我觉得就很明显了。

def load_teststep(raw_testinfo):
  ...

def load_testcase(raw_testcase):
  ...

def load_testcase_v2(raw_testcase):
  """ load testcase in format version 2. """
  ...

def load_testsuite(raw_testsuite):
  ...

剩下一些类似 __extend_with_api_ref() __extend_with_testcase_ref() 等等的函数，无非就是读取你的测试步骤中调用的扩展 api 或者 testcase，locate_debugtalk_py() 函数作用就更明显了。一系列相关的东西都加载完，最后开始 load_test 进行最后的汇总组合。

def load_tests(path, dot_env_path=None):
    """ load testcases from file path, extend and merge with api/testcase definitions.

    Args:
        path (str): testcase/testsuite file/foler path.
            path could be in 2 types:
                - absolute/relative file path
                - absolute/relative folder path
        dot_env_path (str): specified .env file path

    Returns:
        dict: tests mapping, include project_mapping and testcases.
              each testcase is corresponding to a file.
            {
                "project_mapping": {
                    "PWD": "XXXXX",
                    "functions": {},
                    "env": {}
                },
                "testcases": [
                    {   # testcase data structure
                        "config": {
                            "name": "desc1",
                            "path": "testcase1_path",
                            "variables": [],                    # optional
                        },
                        "teststeps": [
                            # test data structure
                            {
                                'name': 'test desc1',
                                'variables': [],    # optional
                                'extract': [],      # optional
                                'validate': [],
                                'request': {}
                            },
                            test_dict_2   # another test dict
                        ]
                    },
                    testcase_2_dict     # another testcase dict
                ],
                "testsuites": [
                    {   # testsuite data structure
                        "config": {},
                        "testcases": {
                            "testcase1": {},
                            "testcase2": {},
                        }
                    },
                    testsuite_2_dict
                ]
            }

    """
    if not os.path.exists(path):
        err_msg = "path not exist: {}".format(path)
        logger.log_error(err_msg)
        raise exceptions.FileNotFound(err_msg)

    if not os.path.isabs(path):
        path = os.path.join(os.getcwd(), path)

    load_project_tests(path, dot_env_path)
    tests_mapping = {
        "project_mapping": project_mapping
    }

    def __load_file_content(path):
        loaded_content = None
        try:
            loaded_content = load_test_file(path)
        except exceptions.FileFormatError:
            logger.log_warning("Invalid test file format: {}".format(path))

        if not loaded_content:
            pass
        elif loaded_content["type"] == "testsuite":
            tests_mapping.setdefault("testsuites", []).append(loaded_content)
        elif loaded_content["type"] == "testcase":
            tests_mapping.setdefault("testcases", []).append(loaded_content)
        elif loaded_content["type"] == "api":
            tests_mapping.setdefault("apis", []).append(loaded_content)

    if os.path.isdir(path):
        files_list = load_folder_files(path)
        for path in files_list:
            __load_file_content(path)

    elif os.path.isfile(path):
        __load_file_content(path)

    return tests_mapping

总之，多看看源码就好辣 hhh
源码：https://github.com/httprunner/httprunner
回答问题的相关代码文件： ./httprunner/loader.py

你都从 yaml 文件里面拿到了所有的请求和参数这一些，接下来不就像写接口测试脚本一样吗，把对应的参数放到 python 的脚本里面对应的位置，然后用 requests 去执行。所谓的 testsuite testcase teststep ，无非就是执行过程中的先后顺序，然后再去遍历执行每个过程中又嵌套的 testsuite testcase teststep。

""" Running testcases. """
   Examples:
       >>> tests_mapping = {
               "project_mapping": {
                   "functions": {}
               },
               "testcases": [
                   {
                       "config": {
                           "name": "XXXX",
                           "base_url": "http://127.0.0.1",
                           "verify": False
                       },
                       "teststeps": [
                           {
                               "name": "test description",
                               "variables": [],        # optional
                               "request": {
                                   "url": "http://127.0.0.1:5000/api/users/1000",
                                   "method": "GET"
                               }
                           }
                       ]
                   }
               ]
           }

       >>> testcases = parser.parse_tests(tests_mapping)
       >>> parsed_testcase = testcases[0]

       >>> test_runner = runner.Runner(parsed_testcase["config"])
       >>> test_runner.run_test(parsed_testcase["teststeps"][0])

# 应该大致是这样的，可能有不同，只看过1.0的
# create_test 返回具体干活的test方法，testdict_in_yml是解析yml后提取的数据
test_method = create_test(testdict_in_yml)

my_testcase_class = type('MyTestCase', (unittest.TestCase,), {})
setattr(my_testcase_class, test_method_name, test_method)

test_loader = unittest.TestLoader()
loaded_testcase =test_loader.loadTestsFromTestCase(my_testcase_class)
TestSuite.addTest(loaded_testcase)

楼上正解，在 yaml 里面，我们定义了一些数据，比如当前测试的接口名、测试的功能模块名等等，有了这些数据，通过 type 方法，动态生成一个继承自 unittset.TestCase 类的测试类，类名就可以是我们定义的这些数据中的一个，这一步就和我们平时用 unittest 一样的，然后用 setattr 方法，给创建的动态测试类添加测试方法，方法名也可以是我们定义的数据中的一个，最后添加到 suite 中

你的分析好像漏了最重要的一步，怎么把对应的 yaml 映射成一个 test 方法，并且添加进 testsuit 中
在./httprunner/api.py 文件中的这个方法应该才是给对应的配置文件绑定一个测试方法的核心吧。我是没有看懂这个方法的运行原理。。。

def _add_tests(self, testcases):
        """ initialize testcase with Runner() and add to test suite.

        Args:
            testcases (list): testcases list.

        Returns:
            unittest.TestSuite()

        """
        def _add_test(test_runner, test_dict):
            """ add test to testcase.
            """
            def test(self):
                try:
                    test_runner.run_test(test_dict)
                except exceptions.MyBaseFailure as ex:
                    self.fail(str(ex))
                finally:
                    self.meta_datas = test_runner.meta_datas

            if "config" in test_dict:
                # run nested testcase
                test.__doc__ = test_dict["config"].get("name")
                variables = test_dict["config"].get("variables", {})
            else:
                # run api test
                test.__doc__ = test_dict.get("name")
                variables = test_dict.get("variables", {})

            if isinstance(test.__doc__, parser.LazyString):
                try:
                    parsed_variables = parser.parse_variables_mapping(variables)
                    test.__doc__ = parser.parse_lazy_data(
                        test.__doc__, parsed_variables
                    )
                except exceptions.VariableNotFound:
                    test.__doc__ = str(test.__doc__)

            return test

        test_suite = unittest.TestSuite()
        for testcase in testcases:
            config = testcase.get("config", {})
            test_runner = runner.Runner(config)
            TestSequense = type('TestSequense', (unittest.TestCase,), {})

            tests = testcase.get("teststeps", [])
            for index, test_dict in enumerate(tests):
                times = test_dict.get("times", 1)
                try:
                    times = int(times)
                except ValueError:
                    raise exceptions.ParamsError(
                        "times should be digit, given: {}".format(times))

                for times_index in range(times):
                    # suppose one testcase should not have more than 9999 steps,
                    # and one step should not run more than 999 times.
                    test_method_name = 'test_{:04}_{:03}'.format(index, times_index)
                    test_method = _add_test(test_runner, test_dict)
                    setattr(TestSequense, test_method_name, test_method)

            loaded_testcase = self.test_loader.loadTestsFromTestCase(TestSequense)
            setattr(loaded_testcase, "config", config)
            setattr(loaded_testcase, "teststeps", tests)
            setattr(loaded_testcase, "runner", test_runner)
            test_suite.addTest(loaded_testcase)

        return test_suite

ok 仔细研究一下