Compare commits

..

No commits in common. "main" and "v2.0.0" have entirely different histories.
main ... v2.0.0

14 changed files with 152 additions and 281 deletions

View File

@ -3,6 +3,7 @@ LICENSE
*.log *.log
README.md README.md
requirements.txt requirements.txt
test_*
Dockerfile Dockerfile
*docker-test* *docker-test*
*.log *.log
@ -10,12 +11,6 @@ Dockerfile
*.csv *.csv
*.toml *.toml
*.git* *.git*
.dockerignore
.DS_Store
.vscode/*
thunder-tests/*
.drone.yml
# Byte-compiled / optimized / DLL files # Byte-compiled / optimized / DLL files
__pycache__/ __pycache__/
*.py[cod] *.py[cod]

View File

@ -1,55 +0,0 @@
kind: pipeline
name: default
steps:
- name: docker
image: plugins/docker
settings:
registry: git.jonb.io
dry_run: false
username: jblu
password:
from_secret: gittea_drone
repo: git.jonb.io/jblu/qbit-maid
tags:
- latest
when:
branch:
- main
event:
- push
- name: docker-test
image: plugins/docker
settings:
registry: git.jonb.io
dry_run: false
username: jblu
password:
from_secret: gittea_drone
repo: git.jonb.io/jblu/qbit-maid
tags:
- dev
when:
branch:
- dev*
event:
- push
- name: test-main
image: git.jonb.io/jblu/qbit-maid:latest
commands:
- python test_qbitmaid.py
- python test_write_csv.py
when:
branch:
- main
event:
- push
- name: test-dev
image: git.jonb.io/jblu/qbit-maid:dev
commands:
- python test_qbitmaid.py
- python test_write_csv.py
when:
branch:
- dev*
event:
- push

View File

@ -1,8 +1,8 @@
FROM python:alpine3.18 FROM python:alpine3.18
WORKDIR / WORKDIR /
COPY . opt COPY . opt
RUN apk add --no-cache supercronic
RUN pip install requests RUN pip install requests
RUN pip install qbittorrent-api RUN pip install qbittorrent-api
RUN crontab /opt/crontab
RUN chmod +x /opt/entrypoint.sh RUN chmod +x /opt/entrypoint.sh
CMD ["/opt/entrypoint.sh"] CMD ["/opt/entrypoint.sh"]

174
README.md
View File

@ -1,8 +1,6 @@
# qbit-maid # qbit-maid
> Warning: This application removes torrents that are over the minimum age and that are not part of the ignored categories, domains or tags. Please use the delete_torrents feature set to false when first testing its functionality.
[![Build Status](https://drone.jonb.io/api/badges/jblu/qbit-maid/status.svg?ref=refs/heads/main)](https://drone.jonb.io/jblu/qbit-maid) ## Warning: This application removes torrents that are over the minimum age and that are not part of the ignored categories, domains or tags. Please use the delete_torrents feature set to false when first testing its functionality.
The objective is to remove torrents based on the following criteria: The objective is to remove torrents based on the following criteria:
- tracker domain name - tracker domain name
@ -10,92 +8,94 @@ The objective is to remove torrents based on the following criteria:
- ratio - ratio
- state - state
## Install ```mermaid
### Docker(Recommended) graph TD;
qbit-maid.py-->qlogging.py;
[package](https://git.jonb.io/jblu/-/packages/container/qbit-maid/latest) qbit-maid.py-->qlist.py;
qbit-maid.py-->qprocess.py;
docker pull git.jonb.io/jblu/qbit-maid:latest qlogging.py-->qbit-maid.py;
qlist.py-->qbit-maid.py;
#### Docker Run Command: qprocess.py-->qbit-maid.py;
> Please note it is best practice to escape spaces in variables. That is why there is backslashes in the cron schedule.
docker run --name qbit-maid -v /opt/qbit-maid:/config/ -e CRON=0\ 1\ *\ *\ * -e toml_path=/config/config.toml git.jonb.io/jblu/qbit-maid
#### Docker Compose
``` ```
version: '3.3'
services: | File | Purpose |
qbit-maid: | --- | --- |
image: git.jonb.io/jblu/qbit-maid | qbit-maid.py | Client to the qbit api and calls functions from the other files |
container_name: qbit-maid | qlist.py | Builds out torrent lists |
volumes: | qlogging.py | Logging and push notification communication |
- /opt/qbit-maid:/config | qprocess.py | Submits qualifying torrents for deletion |
environment: | test_qbitmaid.py | Unit tests |
- CRON="0 1 * * *" | ignored_categories.json | whitelist for categorys to ignore |
- toml_path=/config/config.toml | ignored_tags.json | whitelist for torrent tags to ignore |
| ignored_trackers.json | whitelist of fqdn names to ignore |
You will need a config.json in the root directory.
| Key | Value |
| --- | --- |
| host | string, ip or hostname of qbittorrent server |
| port | number, port of admin gui(used for api aswell) |
| username | admin account for qbittorrent |
| password | password for admin account |
| log_level | is what log messages are written to the log file. INFO or DEBUG are valid entries(case sensitive) |
| protected_tag | used to mark torrents to handle with care |
| non_protected_tag | we don't care about these torrents |
| log_path | will write a log in root directory if left as is other wise specify other path using forward slashes |
| age | number, seconds for how long we keep torrents from IPTORRENTS |
| minimum_age | age in seconds torrents should reached before they are removed |
| use_pushover | true or false to enable or disable pushover notification summary |
| use_log | true or false to enable or disable writing to alog file |
| po_key | pushover key |
| po_token | pushover api token |
| delete_torrents | true or false to enable or disable deletion. Useful for dry-runs |
| enable_dragnet | true or false to enable dragnet functionality. Useful for debugging |
It should look something like this:
Config.json
``` ```
### Via Git {
git clone https://git.jonb.io/jblu/qbit-maid.git "host": "192.168.1.1",
"port": 8080,
Qbit-maid will look for an environment variable *toml_path* for its configuration.If it doesn't find it, it will look for a config.toml file in it's own directory. "username": "admin",
##### config.toml "password": "admin",
"loglevel": "INFO",
"logpath": "./qc.log",
"protected_tag": "ipt",
"non_protected_tag": "public",
"age": 2419200,
"minimum_age": 432000,
"use_pushover": false,
"use_log": true,
"po_key": "",
"po_token": "",
"delete_torrents": false
"enable_dragnet": false,
"dragnet_outfile": "./orphaned.csv"
}
``` ```
[qbittorrent]
host = "192.168.x.x"
port = 8080
username = "user"
password = "pass"
[logging] You will need a ignored_categories.json in the root directory. This will ignore any of the categories found in the values of the entries.
use_log = true ```
log_level = "INFO" {
log_path = "./qc.log" "example": "general",
"example2": "sonarr"
[app_tags] }
protected_tag = "ipt" ```
non_protected_tag = "public"
You will need a ignored_domains.json in the root directory. This will ignore any torrents from these trackers.
[torrent] ```
age = 2419200 {
minimum_age = 432000 "iptorrents-empirehost": "ssl.empirehost.me",
delete_torrents = false "iptorrents-stackoverflow": "localhost.stackoverflow.tech",
"iptorrents-bgp": "routing.bgp.technology"
[pushover] }
use_pushover = false ```
po_key = "<key>>"
po_token = "<token>>" You will need a ignored_tags.json in the root directory. This will ignore any torrents with these tags.
```
[apprise] {
use_apprise = false "first":"first",
host = "192.168.x.x" "second":"second",
port = 8088 "third":"third"
aurls = 'mailto://user:pass@gmail.com' }
[dragnet]
enable_dragnet = false
dragnet_outfile = "./orphaned.csv"
[telemetry]
enable_telemetry = false
telemetry_outfile = "./telemetry.csv"
[ignored_categories]
tech = "tech"
books = "books"
general = "general"
[ignored_domains]
iptorrents-empirehost = "ssl.empirehost.me"
iptorrents-stackoverflow = "localhost.stackoverflow.tech"
iptorrents-bgp = "routing.bgp.technology"
[ignored_tags]
save = "save"
[healthcheck]
use_healthcheck = false
healthcheck_url = "https://example.com/ping/<uuid>>"
``` ```

View File

@ -1,12 +1,12 @@
[qbittorrent] [qbittorrent]
host = "192.168.x.x" host = "192.168.4.11"
port = 8080 port = 8085
username = "user" username = "jman"
password = "pass" password = "nO^touchy@"
[logging] [logging]
use_log = true use_log = true
log_level = "INFO" log_level = "DEBUG"
log_path = "./qc.log" log_path = "./qc.log"
[app_tags] [app_tags]
@ -14,23 +14,22 @@ protected_tag = "ipt"
non_protected_tag = "public" non_protected_tag = "public"
[torrent] [torrent]
max_age = 2419200 age = 2419200
min_age = 432000 minimum_age = 432000
delete_torrents = false delete_torrents = true
[pushover] [pushover]
use_pushover = false use_pushover = false
po_key = "<key>>" po_key = ""
po_token = "<token>>" po_token = ""
[apprise] [apprise]
use_apprise = false
host = "192.168.x.x" host = "192.168.x.x"
port = 8088 port = 8088
aurls = 'mailto://user:pass@gmail.com' aurls = 'mailto://user:pass@gmail.com'
[dragnet] [dragnet]
enable_dragnet = false enable_dragnet = true
dragnet_outfile = "./orphaned.csv" dragnet_outfile = "./orphaned.csv"
[ignored_categories] [ignored_categories]
@ -44,8 +43,6 @@ iptorrents-stackoverflow = "localhost.stackoverflow.tech"
iptorrents-bgp = "routing.bgp.technology" iptorrents-bgp = "routing.bgp.technology"
[ignored_tags] [ignored_tags]
save = "save" first = "first"
second = "second"
[healthcheck] third = "third"
use_healthcheck = true
healthcheck_url = "https://example.com/ping/<uuid>>"

1
crontab Executable file
View File

@ -0,0 +1 @@
0 1 * * * . /etc/environment; python /opt/qbit-maid.py >> /logfile

View File

@ -1,7 +1,5 @@
#!/bin/sh #!/bin/sh
CRON_CONFIG_FILE="/opt/crontab" printenv | grep -v "no_proxy" >> /etc/environment
echo "${CRON} python /opt/qbit-maid.py" > $CRON_CONFIG_FILE crond -f
exec supercronic -passthrough-logs -quiet $CRON_CONFIG_FILE

View File

@ -12,18 +12,16 @@ from collections import Counter
import csv import csv
import requests as r import requests as r
import os import os
r.packages.urllib3.disable_warnings() import sys
class Qbt: class Qbt:
def __init__(self): def __init__(self):
"""Main object, should be calling functions from qlist.py, qlogging.py and qprocess.py""" """Main object, should be calling functions from qlist.py, qlogging.py and qprocess.py"""
# Open the config. Needs a json file with the data in config.json.example # Open the config. Needs a json file with the data in config.json.example
self.st = datetime.datetime.now() self.st = datetime.datetime.now()
if os.getenv("toml_path"): config_file_path=os.environ["toml_path"]
config_file_path=os.getenv("toml_path") with open(config_file_path, 'rb') as c:
with open(config_file_path, 'rb') as c: self.config = load(c)
self.config = load(c)
if os.path.exists('./config.toml'): if os.path.exists('./config.toml'):
config_file_path = './config.toml' config_file_path = './config.toml'
with open(config_file_path, 'rb') as c: with open(config_file_path, 'rb') as c:
@ -55,8 +53,8 @@ class Qbt:
#torrent #torrent
self.delete_torrents = self.config["torrent"]["delete_torrents"] self.delete_torrents = self.config["torrent"]["delete_torrents"]
self.min_age = self.config["torrent"]["min_age"] self.minimum_age = self.config["torrent"]["minimum_age"]
self.max_age = self.config["torrent"]["max_age"] self.age = self.config["torrent"]["age"]
#pushover #pushover
self.use_pushover = self.config["pushover"]["use_pushover"] self.use_pushover = self.config["pushover"]["use_pushover"]
@ -73,10 +71,6 @@ class Qbt:
self.enable_dragnet = self.config["dragnet"]["enable_dragnet"] self.enable_dragnet = self.config["dragnet"]["enable_dragnet"]
self.dragnet_outfile = self.config["dragnet"]["dragnet_outfile"] self.dragnet_outfile = self.config["dragnet"]["dragnet_outfile"]
#telemetry
self.enable_telemetry = self.config["telemetry"]["enable_telemetry"]
self.telemetry_outfile = self.config["telemetry"]["telemetry_outfile"]
#ignored_categories #ignored_categories
self.cat_whitelist = self.config["ignored_categories"] self.cat_whitelist = self.config["ignored_categories"]
@ -86,19 +80,10 @@ class Qbt:
#ignored_tags #ignored_tags
self.ignored_tags = self.config["ignored_domains"] self.ignored_tags = self.config["ignored_domains"]
#healthcheck
self.use_healthcheck = self.config["healthcheck"]["use_healthcheck"]
self.healthcheck_url = self.config["healthcheck"]["healthcheck_url"]
# Calling log and notify functions # Calling log and notify functions
tor_log(self) tor_log(self)
tor_notify(self) tor_notify(self)
self.t = time self.t = time
#start healthcheck job
if self.use_healthcheck:
send_ping(self, r, self.healthcheck_url.rstrip("/") + "/start" )
# Pulling domain names to treat carefully # Pulling domain names to treat carefully
self.tracker_list = [] self.tracker_list = []
self.up_tor_counter = 0 self.up_tor_counter = 0
@ -121,7 +106,6 @@ class Qbt:
if self.use_log: if self.use_log:
list_qbit_api_info(self) list_qbit_api_info(self)
list_first_tor(self) list_first_tor(self)
debug_torrent_list(self)
build_tor_list(self) build_tor_list(self)
process_counts(self) process_counts(self)
if self.use_log: if self.use_log:
@ -137,8 +121,6 @@ class Qbt:
tor_notify_summary(self) tor_notify_summary(self)
if self.use_apprise: if self.use_apprise:
tor_notify_apprise(self, r, apprise_notify) tor_notify_apprise(self, r, apprise_notify)
if self.use_healthcheck:
send_ping(self, r, self.healthcheck_url)
# Run # Run
if __name__== "__main__": if __name__== "__main__":
Qbt() Qbt()

View File

@ -12,7 +12,7 @@ def build_tor_list(self):
# Need a way to tag when the tracker is blank # Need a way to tag when the tracker is blank
if is_tracker_blank(torrent['tracker']): if is_tracker_blank(torrent['tracker']):
if self.use_log: if self.use_log:
self.tl.warning(f'Torrent doesn\'t have a tracker ["{torrent["name"][0:20]}..."] hash: {torrent["hash"]}') self.tl.warning(f'Torrent doesn\'t have a tracker ["{torrent["name"][0:20]}..."] [{torrent["tracker"]}]hash: {torrent["hash"]}')
self.ignored_counter += 1 self.ignored_counter += 1
continue continue
elif is_cat_ignored(torrent['category'], self.cat_whitelist.values()): elif is_cat_ignored(torrent['category'], self.cat_whitelist.values()):
@ -32,9 +32,9 @@ def build_tor_list(self):
self.qbt_client.torrents_add_tags(self.tracker_protected_tag,torrent['hash']) self.qbt_client.torrents_add_tags(self.tracker_protected_tag,torrent['hash'])
elif is_not_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()): elif is_not_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
self.qbt_client.torrents_add_tags(self.tracker_non_protected_tag,torrent['hash']) self.qbt_client.torrents_add_tags(self.tracker_non_protected_tag,torrent['hash'])
if is_preme(torrent['seeding_time'], self.min_age): if is_preme(torrent['added_on'], self.minimum_age, self.t.time()):
self.preme_tor_counter += 1 self.preme_tor_counter += 1
self.tl.debug(f'Premature torrent: ["{torrent["name"][0:20]}..."] Seconds Seeded: [{torrent["seeding_time"]}] hash: {torrent["hash"]}') self.tl.debug(f'Premature torrent: ["{torrent["name"][0:20]}..."] hash: {torrent["hash"]}')
continue continue
elif is_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()): elif is_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
if is_tag_blank(torrent['tags']): if is_tag_blank(torrent['tags']):
@ -50,8 +50,8 @@ def build_tor_list(self):
self.tracker_list.append(torrent) self.tracker_list.append(torrent)
def is_preme(seeding_time, minage): def is_preme(added, minage, time):
if seeding_time <= minage: if added + minage >= time:
return True return True
def is_cat_ignored(cat, catlist): def is_cat_ignored(cat, catlist):

View File

@ -85,12 +85,3 @@ def get_script_runtime(self):
self.extm = f"Execution time: [{elapsed_time}]" self.extm = f"Execution time: [{elapsed_time}]"
if self.use_apprise: if self.use_apprise:
self.extm = f"Execution time: [{elapsed_time}]" self.extm = f"Execution time: [{elapsed_time}]"
def send_ping(self, req_obj, healthcheck_url):
try:
req_obj.get(healthcheck_url, timeout=5)
except req_obj.RequestException as e:
self.tl.info(f"Ping failed: {e}")
def debug_torrent_list(self):
self.tl.debug(self.torrent_list)

View File

@ -3,10 +3,6 @@ def tor_processor(self):
If torrent meets criteria for deletion, its infohash_v1 will be appended to self.torrent_hash_delete_list If torrent meets criteria for deletion, its infohash_v1 will be appended to self.torrent_hash_delete_list
""" """
for canidate in self.tracker_list: for canidate in self.tracker_list:
if self.enable_telemetry:
header = ['state','ratio','tags','added','hash','name','tracker']
row = [canidate['state'],canidate['ratio'],canidate["tags"],canidate['added_on'],canidate['infohash_v1'],canidate["name"][0:20],canidate['tracker']]
write_csv(self.cv,self.telemetry_outfile,header,row)
if self.use_log: if self.use_log:
self.tl.debug(f'---Reviewing canidate: ["{canidate["name"][0:20]}..."] {canidate["infohash_v1"]}---') self.tl.debug(f'---Reviewing canidate: ["{canidate["name"][0:20]}..."] {canidate["infohash_v1"]}---')
if is_downloading(canidate['state']): if is_downloading(canidate['state']):
@ -16,9 +12,9 @@ def tor_processor(self):
elif is_protected_under_ratio(canidate['ratio'], 1.05, self.tracker_protected_tag, canidate["tags"]): elif is_protected_under_ratio(canidate['ratio'], 1.05, self.tracker_protected_tag, canidate["tags"]):
if self.use_log: if self.use_log:
self.tl.debug(f'["{canidate["name"][0:20]}..."] is below a 1.05 ratio({canidate["ratio"]})') self.tl.debug(f'["{canidate["name"][0:20]}..."] is below a 1.05 ratio({canidate["ratio"]})')
if is_old_tor(canidate['time_active'], self.max_age): if is_old_tor(canidate['added_on'], self.age, self.t.time()):
if self.use_log: if self.use_log:
self.tl.debug(f'["{canidate["name"][0:20]}..."] Seconds old: {canidate["time_active"]} Delta: {canidate["time_active"] - self.max_age}') self.tl.debug(f'["{canidate["name"][0:20]}..."] Seconds old: {self.t.time() - self.age - canidate["added_on"]}')
self.torrent_hash_delete_list.append(canidate['infohash_v1']) self.torrent_hash_delete_list.append(canidate['infohash_v1'])
if self.use_log: if self.use_log:
self.tl.info(f'Submitted ["{canidate["name"][0:20]}..."] for deletion from the protected list.') self.tl.info(f'Submitted ["{canidate["name"][0:20]}..."] for deletion from the protected list.')
@ -34,9 +30,7 @@ def tor_processor(self):
self.tl.info(f'Submitted ["{canidate["name"][0:20]}..."] for deletion.') self.tl.info(f'Submitted ["{canidate["name"][0:20]}..."] for deletion.')
else: else:
if self.enable_dragnet: if self.enable_dragnet:
header = ['state','ratio','tags','added','thash','tname','trname'] dragnet(self.cv,self.dragnet_outfile,canidate['state'],canidate['ratio'],canidate["tags"],canidate['added_on'],self.age,self.t.time(),canidate['infohash_v1'],canidate["name"][0:20],canidate['tracker'])
row = [canidate['state'],canidate['ratio'],canidate["tags"],canidate['added_on'],canidate['infohash_v1'],canidate["name"][0:20],canidate['tracker']]
write_csv(self.cv,self.dragnet_outfile,header,row)
self.tl.info(f'["{canidate["name"][0:20]}..."] is orphaned.') self.tl.info(f'["{canidate["name"][0:20]}..."] is orphaned.')
self.up_tor_counter += 1 self.up_tor_counter += 1
continue continue
@ -62,8 +56,8 @@ def is_protected_under_ratio(torratio, setratio, settag, tortag):
if torratio < float(setratio) and settag in tortag: if torratio < float(setratio) and settag in tortag:
return True return True
def is_old_tor(realage, maxage): def is_old_tor(toradd, setage, currenttime):
if realage >= maxage: if toradd + setage <= currenttime:
return True return True
def is_protected_over_ratio(torratio, setratio, settag, tortag): def is_protected_over_ratio(torratio, setratio, settag, tortag):
@ -74,7 +68,9 @@ def is_not_protected_tor(setnonprotectedtag, tortags):
if setnonprotectedtag in tortags: if setnonprotectedtag in tortags:
return True return True
def write_csv(csv_obj,outfile,header,row): def dragnet(csv_obj,outfile,state,ratio,tags,added,age,time,thash,tname,trname):
header = ['state','ratio','tags','added','age','time','thash','tname','trname']
row = [state,ratio,tags,added,age,time,thash,tname,trname]
with open(outfile, 'a+', encoding='UTF8', newline='') as f: with open(outfile, 'a+', encoding='UTF8', newline='') as f:
writer = csv_obj.writer(f) writer = csv_obj.writer(f)
if f.tell() == 0: if f.tell() == 0:

21
test_dragnet.py Executable file
View File

@ -0,0 +1,21 @@
from qprocess import dragnet
import csv
import unittest
class TestDragnet(unittest.TestCase):
def test_dragnet_sanity(self):
self.cv = csv
outfile = './test_outfile.csv'
state = 'downloading'
ratio = 1.05
tags = 'ipt'
added = 1
age = 240000
time = 123456
thash = 'asfasdf23412adfqwer'
tname = 'thisismynamehahahah'
trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce'
dragnet(self.cv,outfile,state,ratio,tags,added,age,time,thash,tname,trname)
if __name__ == '__main__':
unittest.main()

View File

@ -1,14 +1,11 @@
import unittest import unittest
import requests as r
from qlist import is_preme,is_cat_ignored,is_tracker_blank,is_protected_tracker,is_not_protected_tracker,is_tag_blank,is_ignored_tag from qlist import is_preme,is_cat_ignored,is_tracker_blank,is_protected_tracker,is_not_protected_tracker,is_tag_blank,is_ignored_tag
from qprocess import is_downloading,is_protected_under_ratio,is_old_tor,is_protected_over_ratio,is_not_protected_tor from qprocess import is_downloading,is_protected_under_ratio,is_old_tor,is_protected_over_ratio,is_not_protected_tor
from qlogging import send_ping
class TestQbitmaid(unittest.TestCase): class TestQbitmaid(unittest.TestCase):
def test_ispreme_sanity(self): def test_ispreme_sanity(self):
self.assertTrue(is_preme(1,1)) self.assertTrue(is_preme(1,1,1))
self.assertTrue(is_preme(1,2)) self.assertFalse(is_preme(1,1,3))
self.assertFalse(is_preme(2,1))
def test_ispreme(self): def test_ispreme(self):
pass pass
@ -18,7 +15,7 @@ class TestQbitmaid(unittest.TestCase):
self.assertTrue(is_cat_ignored('b', ['a','b','c'])) self.assertTrue(is_cat_ignored('b', ['a','b','c']))
self.assertTrue(is_cat_ignored('c', ['a','b','c'])) self.assertTrue(is_cat_ignored('c', ['a','b','c']))
self.assertFalse(is_cat_ignored('d', ['a','b','c'])) self.assertFalse(is_cat_ignored('d', ['a','b','c']))
self.assertFalse(is_cat_ignored(1, ['a','b','d'])) self.assertFalse(is_cat_ignored(1, ['a','b','c']))
self.assertFalse(is_cat_ignored(1.0000000, ['a','b','c'])) self.assertFalse(is_cat_ignored(1.0000000, ['a','b','c']))
def test_iscatignored(self): def test_iscatignored(self):
@ -71,12 +68,12 @@ class TestQbitmaid(unittest.TestCase):
pass pass
def test_isoldtor_sanity(self): def test_isoldtor_sanity(self):
self.assertFalse(is_old_tor(1,2)) self.assertTrue(is_old_tor(1,2,4))
def test_isoldtor(self): def test_isoldtor(self):
self.assertTrue(is_old_tor(1,1)) self.assertFalse(is_old_tor(1661150664,2419200,1662049004.2101078))
self.assertTrue(is_old_tor(2,1)) self.assertFalse(is_old_tor(1661150664,2419200,1662049004))
self.assertFalse(is_old_tor(1,2)) self.assertFalse(is_old_tor(1661150664.000000,2419200.0000000,1662049004.2101078))
def test_isprotectedoverratio_sanity(self): def test_isprotectedoverratio_sanity(self):
self.assertTrue(is_protected_over_ratio(2,1,'a','a,b,c')) self.assertTrue(is_protected_over_ratio(2,1,'a','a,b,c'))
@ -96,17 +93,6 @@ class TestQbitmaid(unittest.TestCase):
def test_isignoredtag(self): def test_isignoredtag(self):
self.assertTrue(is_ignored_tag(['save'], 'save,public,ipt')) self.assertTrue(is_ignored_tag(['save'], 'save,public,ipt'))
def test_sendpingstart_sanity(self):
send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb" + "/start")
url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb"
send_ping(self, r, url)
def test_sendping_start(self):
url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb/"
send_ping(self, r, url.strip("/") + "/start")
send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb")
# def test__sanity(self): # def test__sanity(self):
# pass # pass

View File

@ -1,41 +0,0 @@
from qprocess import write_csv
import csv
import unittest
import os
# Torrent Items needed for testing
# canidate['state'] {canidate["ratio"]} {torrent["tags"]} torrent["seeding_time"] {torrent["category"]} {torrent["name"][0:20]} {canidate["time_active"]}
class TestWriteCSV(unittest.TestCase):
def test_write_csv_dragnet(self):
self.cv = csv
outfile = './test_dragnet_outfile.csv'
state = 'downloading'
ratio = 1.05
tags = 'ipt'
added = 1
age = 240000
time = 123456
thash = 'asfasdf23412adfqwer'
tname = 'thisismynamehahahah'
trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce'
header = ['state','ratio','tags','added','age','time','thash','tname','trname']
row = [state,ratio,tags,added,age,time,thash,tname,trname]
write_csv(self.cv,outfile,header,row)
self.assertTrue(os.path.exists(outfile))
def test_write_csv_telemetry(self):
self.cv = csv
outfile = './test_telemetry_outfile.csv'
state = 'downloading'
ratio = 1.05
tags = 'ipt'
added = 1
thash = 'asfasdf23412adfqwer'
tname = 'thisismynamehahahah'
trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce'
header = ['state','ratio','tags','added','hash','name','tracker']
row = [state,ratio,tags,added,thash,tname,trname]
write_csv(self.cv,outfile,header,row)
self.assertTrue(os.path.exists(outfile))
if __name__ == '__main__':
unittest.main()