因為筆記本不是真正的文件,許多現有的測試庫可能不使用筆記本中的代碼。有兩種方法:
- 顯式地創建和執行測試套件
- 使用納特庫對於Python
(實際測試的火花代碼我們可以使用庫中提到這一點回答)
使代碼可測試的第一步是要有一個正確的代碼組織——這是常見的,筆記本有線性函數編寫的代碼,沒有顯式的函數或類。在筆記本測試的代碼,我們需要把代碼分成幾個筆記本:
- 創建一個單獨的筆記本你想要測試的功能
- 為測試創建一個單獨的筆記本
- 創建一個單獨的筆記本對你的入口點(“主要筆記本”)
顯式地創建和執行測試套件
筆記本的功能,你想測試(第一項)需要列入第二和第三筆記本使用運行%指令。
在筆記本測試:
- 包括你最喜歡的測試框架
- 定義測試類或測試函數
- 創建一個測試套件,它可以顯式地創建或我們可以使用Python環境中查找並生成這個測試sute
- 執行測試套件
例如,如果我有一個筆記本的功能generate_data和get_data_prediction筆記本名稱“LibraryFunctions”我可以定義一個測試(我們需要“% LibraryFunctions運行”在此之前測試):
進口unittest類SimpleTest (unittest.TestCase): def test_data_generation(自我):n = 100 name = " tmp42 generate_data (n = n, name =名稱)df =火花。sql (f“select * from{名稱}”)self.assertEqual (df.count (), n) def test_data_prediction(自我):預測= get_data_prediction()的自我。42 assertEqual(預測)
我可以顯式地生成測試套件,但這很麻煩:
def generate_test_class_suite():套件= unittest.TestSuite () suite.addTest (SimpleTest (test_data_generation)) suite.addTest (SimpleTest (test_data_prediction))回歸套件test_suite = generate_test_class_suite ()
但它會更容易使用輔助函數自動發現所有測試用例:
def discover_test_classes (): test_classes = [obj的名字,obj globals () . items()如果name.endswith(測試)和obj。__module__ = =的__main__ '和isinstance (obj,類型)和unittest.case。TestCase (obj.__bases__)]套件設置= unittest.TestSuite()在test_classes test_class:為測試在unittest.defaultTestLoader.getTestCaseNames (test_class): suite.addTest (test_class(測試))回歸套件test_suite = discover_test_classes ()
然後我們可以使用測試運行器執行我們的測試套件(您可以使用其他跑步者,從unittest-xml-reporting圖書館XMLTestRunner如果你想生成JUnit測試結果的XML文件):
跑步者= unittest.TextTestRunner()結果= runner.run(套房)
(請注意,如果你想重新運行測試,您可能需要重新生成測試套件)
使用納特庫
Nutter圖書館是由微軟專為磚筆記本的單元測試。它支持以下功能:
- 可以交互式地執行測試,安排工作,或者通過命令行觸發
- 它可以自動發現所有筆記本測試(包括子文件夾),並執行
- 代碼分為/運行維護階段,可選之前/之後調用——你需要遵循命名約定!例如,您需要定義函數run_ <名稱>調用測試函數,有相應的函數assertion_ <名稱>應該檢查執行結果
- 實際檢查完成這樣的框架,Chispa
- 可以並行執行的測試(但要小心)
- 它可以發布JUnit格式的結果
- 很容易融入Azure DevOps &其他CI / CD係統
你仍然需要功能分割成一個單獨的筆記本,包括成一個筆記本,你使用Nutter定義測試。這是很小的例子如何定義測試使用納特,並執行它(它從命令行交互和觸發或CI / CD管道):
從運行時。nutterfixture進口nutterfixture,標簽類Test1Fixture (nutterfixture): def __init__(自我):自我。code2_table_name = " my_data "自我。code1_view_name = " my_cool_data "自我。code1_num_entries = 100 NutterFixture.__init__(自我)def run_name1(自我):generate_data1 (n =自我。code1_num_entries name = self.code1_view_name) def assertion_name1(自我):df = spark.read.table (self.code1_view_name)斷言(df.count () = = self.code1_num_entries)結果= Test1Fixture () .execute_tests()打印(result.to_string ()) is_job = dbutils.notebook.entry_point.getDbutils () .notebook () .getContext () .currentRunId () .isDefined()如果is_job: result.exit (dbutils)
完整的端到端示例,包括說明如何設置CI / CD管道Azure DevOps能找到這個存儲庫。
