Fix: deprication warnings, TypeError (#7)

## Description

- replaced 'cgi' with 'EmailMessage' due to deprication in Python3.13;
- fixed a TypeError: "a bytes-like object is required, not 'str'";
- fixed xml parsing;
- use 'unittest' instead of custom framework;
- removed Python 3.2 from tests.yaml;

## Testing

By @dnzbk:
 - NZBGet v23.1/Windows 11;

.github/workflows/tests.yml

@@ -10,5 +10,5 @@ jobs:
   tests:
     uses: nzbgetcom/nzbget-extensions/.github/workflows/python-tests.yml@main
     with:
-      python-versions: "3.2 3.3 3.4 3.5 3.6 3.7 3.8 3.9 3.10 3.11 3.12"
+      python-versions: "3.3 3.4 3.5 3.6 3.7 3.8 3.9 3.10 3.11 3.12"
       supported-python-versions: "3.9 3.10 3.11 3.12"

README.md

@@ -3,9 +3,8 @@
 
 ## NZBGet Versions
 
-- pre-release v23+  [v3.0](https://github.com/nzbgetcom/Extension-FailureLink/releases/tag/v3.0)
-- stable  v22 [v2.0](https://github.com/nzbgetcom/Extension-FailureLink/releases/tag/v2.0)
-- legacy  v21 [v2.0](https://github.com/nzbgetcom/Extension-FailureLink/releases/tag/v2.0)
+- pre-release v23+ [v3.1](https://github.com/nzbgetcom/Extension-FailureLink/releases/tag/v3.1)
+- stable v22 [v2.0](https://github.com/nzbgetcom/Extension-FailureLink/releases/tag/v2.0)
 
 > **Note:** This script is compatible with python 3.8.x and above.
 If you need support for Python 2.x or older Python3.x versions please use [v1.21](https://github.com/nzbgetcom/Extension-FailureLink/releases/tag/v1.21) release.

main.py

@@ -27,13 +27,14 @@
 import subprocess
 import json
 import ssl
-import cgi
 import shutil
 import stat
 from base64 import standard_b64encode
 from xmlrpc.client import ServerProxy
 import urllib.request, urllib.error
 from urllib.error import HTTPError
+from email.message import EmailMessage
+import xml.etree.ElementTree as ET
 
 # Exit codes used by NZBGet
 POSTPROCESS_SUCCESS=93
@@ -76,11 +77,11 @@
         FFPROBE = os.path.join(PROGRAM_DIR, 'avprobe')
     else:
         try:
-            FFPROBE = subprocess.Popen(['which', 'ffprobe'], stdout=subprocess.PIPE).communicate()[0].strip()
+            FFPROBE = subprocess.Popen(['which', 'ffprobe'], stdout=subprocess.PIPE).communicate()[0].decode('utf-8').strip()
         except: pass
         if not FFPROBE: 
             try:
-                FFPROBE = subprocess.Popen(['which', 'avprobe'], stdout=subprocess.PIPE).communicate()[0].strip()
+                FFPROBE = subprocess.Popen(['which', 'avprobe'], stdout=subprocess.PIPE).communicate()[0].decode('utf-8').strip()
             except: pass
 if CHECKVIDEO and FFPROBE:
     result = 1
@@ -237,7 +238,7 @@ def connectToNzbGet():
 
 	# Build an URL for XML-RPC requests
 	# TODO: encode username and password in URL-format
-	rpcUrl = 'http://%s:%s@%s:%s/xmlrpc' % (username, password, host, port);
+	rpcUrl = 'http://%s:%s@%s:%s/xmlrpc' % (username, password, host, port)
 
 	# Create remote server object
 	nzbget = ServerProxy(rpcUrl)
@@ -251,14 +252,19 @@ def queueNzb(filename, category, nzbcontent64):
 	nzbget.append(filename, category, 0, True, nzbcontent64, True, '', 0, 'ALL')
 
 	# We need to find the id of the added nzb-file
-	groups = nzbget.listgroups()
-	groupid = 0
+	root = ET.fromstring(nzbget.listgroups().data)
+	groups = root.findall('.//struct')
+
 	for group in groups:
+		nzb_id = group.find(".//member[name='NZBID']/value/i4").text
+		nzb_filename = group.find(".//member[name='NZBFilename']/value/string").text
+
 		if verbose:
-			print(group)
-		if group['NZBFilename'] == filename:
-			groupid = group['NZBID']
-			break;
+			print(f'NZBID: {nzb_id}, NZBFilename: {nzb_filename}')
+
+		if nzb_filename == filename:
+			groupid = int(nzb_id)
+			break
 
 	if verbose:
 		print('GroupID: %i' % groupid)
@@ -267,7 +273,7 @@ def queueNzb(filename, category, nzbcontent64):
 
 
 def setupDnzbHeaders(groupid, headers):
-	for header in headers.headers:
+	for header in headers:
 		if verbose:
 			print(header.strip())
 		if header[0:7] == 'X-DNZB-':
@@ -359,13 +365,11 @@ def main():
         print('[INFO] Another release found, adding to queue')
         sys.stdout.flush()
 
-        # Parsing filename from headers
-
-        params = cgi.parse_header(headers.get('Content-Disposition', ''))
-        if verbose:
-                print(params)
+        # Parsing filename from header
+        msg = EmailMessage()
+        msg['Content-Disposition'] = headers.get('Content-Disposition', '')
+        filename = msg.get_filename()
 
-        filename = params[1].get('filename', '')
         if verbose:
                 print('filename: %s' % filename)
 
@@ -392,4 +396,4 @@ def main():
 
 # All OK, returning exit status 'POSTPROCESS_SUCCESS' (int <93>) to let NZBGet know
 # that our script has successfully completed.
-sys.exit(POSTPROCESS_SUCCESS)
+sys.exit(POSTPROCESS_SUCCESS)

manifest.json

@@ -4,7 +4,7 @@
 	"homepage": "https://github.com/nzbgetcom/Extension-FailureLink/",
 	"kind": "POST-PROCESSING",
 	"displayName": "Failure Link",
-	"version": "3.0.0",
+	"version": "3.1",
 	"author": "Andrey Prygunkov",
 	"license": "GNU",
 	"about": "Checks videos to determine if they are corrupt. Inform indexer site about failed or corrupt download and request a replacement nzb.",

test_data/test_response.xml

@@ -1,11 +1,16 @@
-<?xml version="1.0" encoding="utf-8"?>
 <array>
 <data>
 <value>
 <struct>
+<member><name>NZBID</name><value><i4>2</i4></value></member>
+<member><name>NZBFilename</name><value><string>2.nzb</string></value></member>
+</struct>
+</value>
+<value>
+<struct>
 <member><name>NZBID</name><value><i4>1</i4></value></member>
 <member><name>NZBFilename</name><value><string>1.nzb</string></value></member>
 </struct>
 </value>
 </data>
-</array>
+</array>

tests.py

@@ -21,11 +21,12 @@
 import sys
 from os.path import dirname
 import os
-import traceback
 import subprocess
 import http.server
 import xmlrpc
+import xmlrpc.server
 import threading
+import unittest
 
 POSTPROCESS_SUCCESS=93
 POSTPROCESS_NONE=95
@@ -39,14 +40,6 @@
 username = 'TestUser'
 password = 'TestPassword'
 
-def RUN_TESTS():
-    TEST('Should not be executed if nzbget version is incompatible', TEST_COMPATIBALE_NZBGET_VERSION)
-    TEST('Should be success if no failure and no video check ', TEST_NOT_FAILURE_NOT_VIDEOCHECK)
-    TEST('Should be success if no failure link', TEST_NO_FAILURE_LINK)
-    TEST('Should delete tmp dir and if no failure', TEST_DELETE_DIR)
-    TEST('Should check video corruption without ffprobe', TEST_CHECK_VIDEO_CORRUPTION_WITHOUT_FFPROBE)
-    # TODO: Test rpc and download another release
-
 class RequestWithFileId(http.server.BaseHTTPRequestHandler):
 	def do_GET(self):
 		self.send_response(200)
@@ -68,19 +61,24 @@ def do_POST(self):
 		self.wfile.write(response.encode('utf-8'))
 		f.close()
 
-def TEST(statement: str, test_func):
-	print('\n********************************************************')
-	print('TEST:', statement)
-	print('--------------------------------------------------------')
-
-	try:
-		test_func()
-		print(test_func.__name__, '...SUCCESS')
-	except Exception as e:
-		print(test_func.__name__, '...FAILED')
-		traceback.print_exc()
-	finally:
-		print('********************************************************\n')
+class RequestNZBGet(xmlrpc.server.SimpleXMLRPCServer):
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header("Content-Type", "text/plain")
+        self.send_header("Content-Disposition", "attachment;filename=1.nzb")
+        self.send_header("X-DNZB-Category", "movie")
+        self.end_headers()
+        with open(test_data_dir + '/1.nzb', 'rb') as f:
+            self.wfile.write(f.read())
+
+    def do_POST(self):
+        self.log_request()
+        self.send_response(200)
+        self.send_header("Content-Type", "text/xml")
+        self.end_headers()
+        with open(test_data_dir + '/test_response.xml', 'rb') as f:
+            response = xmlrpc.client.dumps((f.read(),), allow_none=False, encoding=None)
+            self.wfile.write(response.encode('utf-8'))
 
 def get_python(): 
 	if os.name == 'nt':
@@ -124,68 +122,69 @@ def set_default_env():
 	os.environ.pop('NZBPO_MEDIAEXTENSIONS', None)
 	os.environ.pop('NZBPO_FFPROBE', None)
 
-
-def TEST_COMPATIBALE_NZBGET_VERSION():
-	os.environ.pop('NZBOP_FEEDHISTORY', None)
-	[out, code, err] = run_script()
-	assert('*** NZBGet post-processing script ***' in out)
-	assert('This script is supposed to be called from nzbget (12.0 or later).' in out)
-	assert(code == POSTPROCESS_ERROR)
-
-def TEST_NOT_FAILURE_NOT_VIDEOCHECK():
-	set_default_env()
-	os.environ['NZBPP_PARSTATUS'] = '0'
-	os.environ['NZBPO_CHECKVID'] = 'no'
-	[out, code, err] = run_script()
-	assert(code == POSTPROCESS_SUCCESS)
-
-def TEST_NO_FAILURE_LINK():
-	set_default_env()
-	os.environ['NZBPP_PARSTATUS'] = '1'
-	os.environ['NZBPO_CHECKVID'] = 'no'
-	[out, code, err] = run_script()
-	assert(code == POSTPROCESS_SUCCESS)
-
-def TEST_DELETE_DIR():
-	set_default_env()
-	os.mkdir(tmp_dir)
-	os.environ['NZBPP_PARSTATUS'] = '1'
-	os.environ['NZBPP_DIRECTORY'] = tmp_dir
-	os.environ['NZBPO_DELETE'] = 'yes'
-	[out, code, err] = run_script()
-	assert(os.path.isdir(tmp_dir) == False)
-	assert(code == POSTPROCESS_SUCCESS)
-
-def TEST_CHECK_VIDEO_CORRUPTION_WITHOUT_FFPROBE():
-	set_default_env()
-	os.environ['NZBPP_PARSTATUS'] = '0'
-	os.environ['NZBPO_DOWNLOADANOTHERRELEASE'] = 'no'
-	os.environ['NZBPR__DNZB_FAILURE'] = 'https://link'
-	os.environ['NZBPO_CHECKVID'] = 'yes'
-	os.environ['NZBPO_MEDIAEXTENSIONS'] = '.mp4'
-	os.environ['NZBPO_TESTVID'] = test_data_dir + '/corrupted.mp4'
-	os.environ['NZBPP_DIRECTORY'] = test_data_dir
-	[out, code, err] = run_script()
-	assert('[WARNING] Failed to locate ffprobe, video corruption detection disabled!' in out)
-	assert('[WARNING] Install ffmpeg with x264 support to enable this feature  ...' in out)
-	assert(code == POSTPROCESS_SUCCESS)
-
-def TEST_DOWNLOAD_ANOTHER_RELEASE():
-	set_default_env()
-	os.environ['NZBPP_PARSTATUS'] = '1'
-	os.environ['NZBPO_DOWNLOADANOTHERRELEASE'] = 'yes'
-	os.environ['NZBPR__DNZB_FAILURE'] = 'http://127.0.0.1:6789'
-	os.environ['NZBPO_CHECKVID'] = 'yes'
-	os.environ['NZBPO_MEDIAEXTENSIONS'] = '.mp4'
-	os.environ['NZBPO_TESTVID'] = test_data_dir + '/corrupted.mp4'
-	os.environ['NZBPP_DIRECTORY'] = test_data_dir
-	server = http.server.HTTPServer((host, int(port)), RequestWithFileId)
-	thread = threading.Thread(target=server.serve_forever)
-	thread.start()
-	[out, code, err] = run_script()
-	server.shutdown()
-	thread.join()
-	assert('[INFO] Requesting another release from indexer site' in out)
-	assert(code == POSTPROCESS_SUCCESS)
-
-RUN_TESTS()
+class Tests(unittest.TestCase):
+	def test_compitable_nzbget_version(self):
+		os.environ.pop('NZBOP_FEEDHISTORY', None)
+		[out, code, err] = run_script()
+		self.assertEqual(code, POSTPROCESS_ERROR)
+
+	def test_not_failure_not_videocheck(self):
+		set_default_env()
+		os.environ['NZBPP_PARSTATUS'] = '0'
+		os.environ['NZBPO_CHECKVID'] = 'no'
+		[out, code, err] = run_script()
+		self.assertEqual(code, POSTPROCESS_SUCCESS)
+
+	def test_no_failure_link(self):
+		set_default_env()
+		os.environ['NZBPP_PARSTATUS'] = '1'
+		os.environ['NZBPO_CHECKVID'] = 'no'
+		[out, code, err] = run_script()
+		self.assertEqual(code, POSTPROCESS_SUCCESS)
+
+	def test_delete_dir(self):
+		set_default_env()
+		os.mkdir(tmp_dir)
+		os.environ['NZBPP_PARSTATUS'] = '1'
+		os.environ['NZBPP_DIRECTORY'] = tmp_dir
+		os.environ['NZBPO_DELETE'] = 'yes'
+		[out, code, err] = run_script()
+		self.assertEqual(os.path.isdir(tmp_dir), False)
+		self.assertEqual(code, POSTPROCESS_SUCCESS)
+
+	def test_check_video_corruption_without_ffprobe(self):
+		set_default_env()
+		os.environ['NZBPP_PARSTATUS'] = '0'
+		os.environ['NZBPO_DOWNLOADANOTHERRELEASE'] = 'no'
+		os.environ['NZBPR__DNZB_FAILURE'] = 'https://link'
+		os.environ['NZBPO_CHECKVID'] = 'yes'
+		os.environ['NZBPO_MEDIAEXTENSIONS'] = '.mp4'
+		os.environ['NZBPO_TESTVID'] = test_data_dir + '/corrupted.mp4'
+		os.environ['NZBPP_DIRECTORY'] = test_data_dir
+		[out, code, err] = run_script()
+		self.assertEqual(code, POSTPROCESS_SUCCESS)
+
+	def test_downlaod_another_release(self):
+		set_default_env()
+		os.environ['NZBPP_PARSTATUS'] = '1'
+		os.environ['NZBPO_DOWNLOADANOTHERRELEASE'] = 'yes'
+		os.environ['NZBPR__DNZB_FAILURE'] = 'http://127.0.0.1:6789'
+		os.environ['NZBPO_CHECKVID'] = 'yes'
+		os.environ['NZBPO_MEDIAEXTENSIONS'] = '.mp4'
+		os.environ['NZBPO_TESTVID'] = test_data_dir + '/corrupted.mp4'
+		os.environ['NZBPP_DIRECTORY'] = test_data_dir
+		httpserver = http.server.HTTPServer((host, int(port)), RequestWithFileId)
+		thread1 = threading.Thread(target=httpserver.serve_forever)
+		thread1.start()
+		rpcserver = RequestNZBGet((host, int(port)))
+		thread2 = threading.Thread(target=rpcserver.serve_forever)
+		thread2.start()
+		[out, code, err] = run_script()
+		httpserver.shutdown()
+		rpcserver.shutdown()
+		thread1.join()
+		thread2.join()
+		self.assertEqual(code, POSTPROCESS_SUCCESS)
+
+if __name__ == '__main__':
+	unittest.main()