如何自动化运行Kusto查询（KQL）文件及实现KQL单元测试自动化？

一、Java（Spring Boot/Quarkus）实现方式 #

1. 读取KQL文件 #

不管是Spring Boot还是Quarkus，都可以通过类路径便捷读取KQL文件：

• Spring Boot 用ResourceLoader加载资源：

@Autowired
private ResourceLoader resourceLoader;

private String loadKqlFile(String filePath) throws IOException {
    Resource resource = resourceLoader.getResource("classpath:kql-tests/" + filePath);
    return Files.readString(Paths.get(resource.getURI()));
}

• Quarkus 直接用Resources类读取：

private String loadKqlFile(String filePath) throws IOException {
    return io.quarkus.runtime.util.Resources.readString("kql-tests/" + filePath);
}

2. 执行KQL查询并验证结果 #

借助Azure Monitor Query Java SDK执行查询并做断言，步骤如下：

1. 添加Maven依赖：

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-monitor-query</artifactId>
    <version>1.2.0</version>
</dependency>
<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-identity</artifactId>
    <version>1.12.0</version>
</dependency>

2. 编写JUnit 5单元测试：

@Test
void testUserLoginKql() throws IOException {
    // 加载KQL文件内容
    String kqlQuery = loadKqlFile("user-login-test.kql");
    
    // 初始化查询客户端
    QueryClient queryClient = new QueryClientBuilder()
        .credential(new DefaultAzureCredentialBuilder().build())
        .buildClient();
    
    // 执行查询（替换为你的工作区ID）
    LogsQueryResult result = queryClient.queryWorkspace("<WORKSPACE_ID>", kqlQuery, new QueryTimeInterval(Duration.ofHours(24)));
    
    // 验证结果：断言返回数据非空、特定字段符合预期
    assertThat(result.getTables().get(0).getRows()).isNotEmpty();
    assertThat(result.getTables().get(0).getRows().get(0).get(0)).isEqualTo("expected-login-count");
}

3. 批量执行测试 #

通过参数化测试遍历指定目录下的所有KQL文件，实现批量自动化：

@ParameterizedTest
@MethodSource("getKqlTestFiles")
void batchRunKqlTests(String fileName) throws IOException {
    String kqlQuery = loadKqlFile(fileName);
    // 复用上述查询与验证逻辑
}

private static Stream<String> getKqlTestFiles() throws IOException {
    return Files.list(Paths.get("src/test/resources/kql-tests"))
        .map(path -> path.getFileName().toString())
        .filter(name -> name.endsWith(".kql"));
}

二、其他自动化方式 #

1. PowerShell 脚本 #

利用Azure PowerShell模块的Invoke-AzOperationalInsightsQuery命令批量执行：

$workspaceId = "<WORKSPACE_ID>"
$kqlFiles = Get-ChildItem -Path "./kql-tests" -Filter "*.kql"

foreach ($file in $kqlFiles) {
    $kqlQuery = Get-Content $file.FullName -Raw
    Write-Host "Running test: $($file.Name)"
    $result = Invoke-AzOperationalInsightsQuery -WorkspaceId $workspaceId -Query $kqlQuery
    
    # 验证结果：检查返回行数是否符合预期
    if ($result.Result.Tables[0].Rows.Count -gt 0) {
        Write-Host "Test $($file.Name) passed"
    } else {
        Write-Host "Test $($file.Name) failed" -ForegroundColor Red
    }
}

2. Python 脚本 #

结合azure-monitor-query库与pytest实现自动化测试：

import os
from azure.monitor.query import LogsQueryClient
from azure.identity import DefaultAzureCredential
import pytest

WORKSPACE_ID = "<WORKSPACE_ID>"
KQL_DIR = "./kql-tests"

@pytest.fixture(scope="session")
def query_client():
    return LogsQueryClient(DefaultAzureCredential())

def get_kql_files():
    return [f for f in os.listdir(KQL_DIR) if f.endswith(".kql")]

@pytest.mark.parametrize("file_name", get_kql_files())
def test_kql_queries(file_name, query_client):
    with open(os.path.join(KQL_DIR, file_name), "r") as f:
        kql_query = f.read()
    
    result = query_client.query_workspace(WORKSPACE_ID, kql_query, timespan="P1D")
    assert len(result.tables[0].rows) > 0, f"Test {file_name} failed: no results returned"

3. CI/CD 集成 #

将脚本集成到Azure DevOps、GitHub Actions等流程，实现触发式或定时自动化：

• GitHub Actions示例（每日执行测试）：

name: KQL Unit Tests
on:
  schedule:
    - cron: '0 0 * * *'
  workflow_dispatch:

jobs:
  run-kql-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install azure-monitor-query azure-identity pytest
      - name: Run tests
        env:
          AZURE_CLIENT_ID: ${{ secrets.AZURE_CLIENT_ID }}
          AZURE_TENANT_ID: ${{ secrets.AZURE_TENANT_ID }}
          AZURE_CLIENT_SECRET: ${{ secrets.AZURE_CLIENT_SECRET }}
        run: pytest kql-tests/ -v

最佳实践 #

• 使用测试专用工作区或隔离数据集，避免影响生产数据
• 对KQL查询进行参数化（如时间范围、过滤条件），提升测试灵活性
• 不仅断言结果行数，还要验证特定字段的取值是否符合预期
• 记录测试结果，生成可视化报告便于排查问题

内容的提问来源于stack exchange，提问作者sam