| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123 |
- # -*- 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 = '<!DOCTYPE html><html>fallback</html>'
- 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 '<!DOCTYPE html>' 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')
|
