# -*- coding:utf-8 -*- from unittest.mock import MagicMock, patch import pytest from dw_base.ds.api import DSClient @pytest.fixture def client(tmp_path): conf = tmp_path / 'ds.ini' conf.write_text( '[dolphinscheduler]\n' 'base_url = http://example/dolphinscheduler\n' 'token = TEST_TOKEN\n', encoding='utf-8', ) return DSClient(conf_path=str(conf)) def test_init_loads_conf(client): assert client.base_url == 'http://example/dolphinscheduler' assert client.token == 'TEST_TOKEN' assert client.session.headers['token'] == 'TEST_TOKEN' def test_init_missing_conf_raises(): with pytest.raises(RuntimeError, match='DS 配置文件不存在'): DSClient(conf_path='/nonexistent/path.ini') def test_init_strips_trailing_slash(tmp_path): conf = tmp_path / 'ds.ini' conf.write_text( '[dolphinscheduler]\n' 'base_url = http://example/dolphinscheduler/\n' 'token = T\n', encoding='utf-8', ) c = DSClient(conf_path=str(conf)) assert c.base_url == 'http://example/dolphinscheduler' def test_get_2xx_returns_json(client): with patch.object(client.session, 'get') as mock_get: resp = MagicMock(status_code=200) resp.json.return_value = {'code': 0, 'data': 'ok'} mock_get.return_value = resp assert client.get('/projects') == {'code': 0, 'data': 'ok'} mock_get.assert_called_once_with( 'http://example/dolphinscheduler/projects', params=None) def test_get_strips_leading_slash(client): """path 带或不带前导 / 都拼成相同 URL。""" with patch.object(client.session, 'get') as mock_get: resp = MagicMock(status_code=200) resp.json.return_value = {} mock_get.return_value = resp client.get('projects') assert mock_get.call_args[0][0] == 'http://example/dolphinscheduler/projects' def test_get_non_2xx_raises(client): with patch.object(client.session, 'get') as mock_get: mock_get.return_value = MagicMock(status_code=500, text='Internal Server Error') with pytest.raises(RuntimeError, match='GET .*failed rc=500'): client.get('/projects') def test_get_2xx_non_json_raises_friendly(client): """新特性：path 错被 SPA fallback 时抛 RuntimeError 含 raw text 前 200 字符。""" with patch.object(client.session, 'get') as mock_get: resp = MagicMock(status_code=200) resp.json.side_effect = ValueError('Expecting value') resp.text = 'fallback' mock_get.return_value = resp with pytest.raises(RuntimeError) as exc: client.get('/wrong/path') msg = str(exc.value) assert '返回非 JSON' in msg assert 'SPA fallback' in msg assert '' in msg def test_get_2xx_non_json_truncates_to_200(client): with patch.object(client.session, 'get') as mock_get: resp = MagicMock(status_code=200) resp.json.side_effect = ValueError() resp.text = 'X' * 500 mock_get.return_value = resp with pytest.raises(RuntimeError) as exc: client.get('/p') # 错误信息含 200 个 X，不含 500 个 assert 'X' * 200 in str(exc.value) assert 'X' * 201 not in str(exc.value) def test_post_2xx_returns_json(client): with patch.object(client.session, 'post') as mock_post: resp = MagicMock(status_code=200) resp.json.return_value = {'ok': True} mock_post.return_value = resp assert client.post('/foo', json_body={'k': 'v'}) == {'ok': True} mock_post.assert_called_once_with( 'http://example/dolphinscheduler/foo', json={'k': 'v'}) def test_post_2xx_non_json_raises_friendly(client): with patch.object(client.session, 'post') as mock_post: resp = MagicMock(status_code=200) resp.json.side_effect = ValueError() resp.text = 'NOT JSON' mock_post.return_value = resp with pytest.raises(RuntimeError, match='POST.*返回非 JSON'): client.post('/foo') def test_post_non_2xx_raises(client): with patch.object(client.session, 'post') as mock_post: mock_post.return_value = MagicMock(status_code=403, text='Forbidden') with pytest.raises(RuntimeError, match='POST .*failed rc=403'): client.post('/foo')