This commit is contained in:
2025-03-03 02:54:33 +03:00
parent 528bb601c5
commit 0898d92fc7
5 changed files with 79 additions and 40 deletions

View File

@@ -1,7 +1,14 @@
from flask import Flask
from .config import Config
def proxy_app():
app = Flask(__name__)
app.config.from_object(Config)
# Подключаем маршруты
from proxy import rss_proxy, healthcheck
rss_proxy.init_proxy(app)
healthcheck.init_healthcheck(app)
return app

View File

@@ -1,10 +1,11 @@
from flask import Response
from proxy import app
def init_healthcheck(app):
@app.route("/health")
def healthcheck():
"""Health check route to monitor service status"""
try:
return Response("OK", status=200)
return Response("healthy", status=200)
except Exception as e:
return f"Error: {e}", 500

View File

@@ -2,10 +2,11 @@ import urllib.parse
from flask import request, Response
import requests
import os
from proxy import app
PROXY_URL = os.getenv("PROXY_URL")
def init_proxy(app):
@app.route("/proxy")
def proxy():
"""Proxy RSS feed with forced re-encoding to UTF-8"""
@@ -19,8 +20,10 @@ def proxy():
proxies = {"http": PROXY_URL, "https": PROXY_URL} if PROXY_URL else None
r = requests.get(url, timeout=10, proxies=proxies)
r.encoding = "windows-1251" if "windows-1251" in r.headers.get("content-type", "").lower() else r.apparent_encoding
response_text = r.text.replace('<?xml version="1.0" encoding="windows-1251"?>', '<?xml version="1.0" encoding="UTF-8"?>')
r.encoding = "windows-1251" if "windows-1251" in r.headers.get("content-type",
"").lower() else r.apparent_encoding
response_text = r.text.replace('<?xml version="1.0" encoding="windows-1251"?>',
'<?xml version="1.0" encoding="UTF-8"?>')
return Response(response_text, content_type="application/xml; charset=utf-8")
except Exception as e:

View File

@@ -1,18 +1,20 @@
import requests
import sys
def check_health(url="http://localhost:5050/health"):
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
print("Health check passed")
sys.exit(0) # Успешная проверка
sys.exit(0)
else:
print(f"Health check failed: {response.status_code}")
sys.exit(1) # Ошибка
sys.exit(1)
except requests.exceptions.RequestException as e:
print(f"Health check failed: {e}")
sys.exit(1) # Ошибка
sys.exit(1)
if __name__ == "__main__":
check_health()

View File

@@ -1,11 +1,12 @@
import unittest
from proxy import app
from unittest.mock import patch, MagicMock
from proxy import proxy_app
class FlaskTestCase(unittest.TestCase):
def setUp(self):
"""Set up for tests: create a test client."""
app = proxy_app()
self.app = app.test_client()
self.app.testing = True
@@ -13,12 +14,37 @@ class FlaskTestCase(unittest.TestCase):
"""Check health endpoint."""
response = self.app.get('/health')
self.assertEqual(response.status_code, 200)
self.assertIn(b'healthy', response.data)
self.assertEqual(response.data, b'healthy')
def test_proxy(self):
"""Test RSS proxying."""
# Sample URL for proxying
url = 'http://example.com/rss'
response = self.app.get(f'/proxy?url={url}')
@patch("proxy.rss_proxy.requests.get")
def test_proxy_success(self, mock_get):
"""Test successful RSS proxying."""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {"content-type": "text/xml; charset=windows-1251"}
mock_response.text = '<?xml version="1.0" encoding="windows-1251"?><rss></rss>'
mock_response.apparent_encoding = "windows-1251"
mock_get.return_value = mock_response
response = self.app.get('/proxy?url=https://example.com/rss.xml')
self.assertEqual(response.status_code, 200)
self.assertIn(b'<?xml', response.data) # Check for XML response
self.assertIn(b'<?xml version="1.0" encoding="UTF-8"?>', response.data) # Проверка перекодировки
@patch("proxy.rss_proxy.requests.get")
def test_proxy_missing_url(self, mock_get):
"""Test proxy request without URL parameter."""
response = self.app.get('/proxy')
self.assertEqual(response.status_code, 400)
self.assertIn(b'Missing URL', response.data)
@patch("proxy.rss_proxy.requests.get")
def test_proxy_request_failure(self, mock_get):
"""Test RSS proxy failure when request fails."""
mock_get.side_effect = Exception("Request failed")
response = self.app.get('/proxy?url=https://example.com/rss.xml')
self.assertEqual(response.status_code, 500)
self.assertIn(b'Error: Request failed', response.data)
if __name__ == '__main__':
unittest.main()