【实例：利用Django管理后台管理IP地址】（五）编写views函数检测每个ip地址占用情况

定制相关数据的展示之后，还需要实现一些功能便于告知用户当前IP地址实际占用情况

需求包括：

1、多线程并发执行对每个IP地址的操作

2、规则：（表字段包括：ip地址、ip使用状态、使用人ID、使用人用户名、备注信息、非法使用天数、无法连接天数）

①若当前IP属于空闲，实际情况：ping通 —> 表示被非法使用，则非法使用天数+1；ping不通 —> 正常情况，因为不被使用，非法使用天数=0，表示已经恢复正常状态

②若当前IP属于使用中，实际情况：ping不通 —> 表示已经不被使用，无法连接天数+1；ping通 —> 正常情况，因为正在使用，无法连接天数=0，表示已经恢复正常状态

③若当前IP属于空闲，若无法连接天数>0，则执行无法连接天数=0，因为无法连接属于正常状况，清除之前的遗留数据；

④若当前IP属于使用中，若非法使用天数>0，则执行非法使用天数=0，因为能连接属于正常状况，清除之前的遗留数据；

所以这里的非法使用天数、无法连接天数都是讲究连续性的，偶尔一两天的不正常状态可能是断电断网/临时借用一两小时，如果连续7天或者20天告警说明是真的没有通知的情况下使用了或者IP已经不被使用了。

views.py

凡是被标注【#把print改成写入日志】的都是当前调试打印出来观察的，之后计划改成写入日志的

from mysiteapp.models import ipaddr_info
# Create your views here.
import subprocess
import threading


# 检查IP是否被占用
def get_ping_result(ip):
    cmd_str = "ping {0} -n 4 -w 600".format(ip)
    DETACHED_PROCESS = 0x00000008   # 不创建cmd窗口
    try:
        subprocess.run(cmd_str, creationflags=DETACHED_PROCESS, check=True)  # 仅用于windows系统
    except subprocess.CalledProcessError as err:
        print("ping不通"+ip)  #把print改成写入日志
        return 0
    else:
        print("ping通"+ip)  # 把print改成写入日志
        return 1

#处理规则
def ip_status_alarm(ip):
    x = get_ping_result(ip)
    try:
        ipobj = ipaddr_info.objects.get(ipaddr=ip)
    except ipaddr_info.DoesNotExist:
        print("Address does not exit!")
    else:
        if ipobj.ipstatus == 0:
            print("0,空闲", ip)  #把print改成写入日志
            if x == 1:
                print("1,ping通", ip)  # 把print改成写入日志
                # 取出connect_alarm_num告警数，+1,存回数据库
                connect_alarm_num0 = ipobj.connect_alarm_num + 1
                ipaddr_info.objects.filter(ipaddr=ip).update(connect_alarm_num=connect_alarm_num0)
            else:
                print("1,ping不通", ip)  # 把print改成写入日志
                ipaddr_info.objects.filter(ipaddr=ip).update(connect_alarm_num=0)
            if ipobj.disconnect_alarm_num > 0:
                # 如果disconnect_alarm_num告警数>0，置为0
                ipaddr_info.objects.filter(ipaddr=ip).update(disconnect_alarm_num=0)
        else:
            print("1,使用中", ip)#把print改成写入日志
            if x == 0:
                print("1,ping不通", ip)  # 把print改成写入日志
                #  取出disconnect_alarm_num告警数，+1
                disconnect_alarm_num1 = ipobj.disconnect_alarm_num + 1
                ipaddr_info.objects.filter(ipaddr=ip).update(disconnect_alarm_num=disconnect_alarm_num1)
            else:
                print("1,ping通", ip)  # 把print改成写入日志
                ipaddr_info.objects.filter(ipaddr=ip).update(disconnect_alarm_num=0)
            if ipobj.connect_alarm_num > 0:
                # 如果connect_alarm_num告警数>0，置为0
                ipaddr_info.objects.filter(ipaddr=ip).update(connect_alarm_num=0)


# 使用多线程
def start_ping(startip, endip):
    stip = startip.split('.')
    enip = endip.split('.')
    tmp_ip = stip
    a = 1
    pthread_list = []
    for i in range(int(stip[3]), int(enip[3]) + 1):
        tmp_ip[3] = str(i)
        ip = '.'.join(tmp_ip)
        print("多线程拼接ip是否成功"+ip)  #改为写入日志
        pthread_list.append(threading.Thread(target=ip_status_alarm, args=(ip,)))
    for item in pthread_list:
        # item.setDaemon(False)
        print("启动第几个线程：", a)  # 改为写入日志
        a = a + 1
        item.start()
    for item in pthread_list:
        item.join()

# 导入Python标准库中的Thread模块
# from threading import Thread
# 创建一个线程，args: 线程执行方法接收的参数，该属性是一个元组，如果只有一个参数也需要在末尾加逗号。
# mthread = threading.Thread(target=function_name, args=(function_parameter1, function_parameterN))
# join ()方法：主线程A中，创建了子线程B，并且在主线程A中调用了B.join()，那么，主线程A会在调用的地方等待，直到子线程B完成操作后，才可以接着往下执行，那么在调用这个线程时可以使用被调用线程的join方法。
# setDaemon()方法。主线程A中，创建了子线程B，并且在主线程A中调用了B.setDaemon(),这个的意思是，把主线程A设置为守护线程，这时候，要是主线程A执行结束了，就不管子线程B是否完成,一并和主线程A退出.这就是setDaemon方法的含义，这基本和join是相反的。
# 此外，还有个要特别注意的：必须在start() 方法调用之前设置，如果不设置为守护线程，程序会被无限挂起。

test.py

比较啰嗦的测试代码，就不精简了。。。。

这里使用了两种测试情况：

① 1.1.1.1是被定义为空闲的IP，但实际可以ping通，而且disconnect_alarm_num（无法连接天数）有残留数据的测试用例

预期结果：connect_alarm_num（非法使用天数）+1、disconnect_alarm_num（无法连接天数）=0

②2.2.2.2是被定义为使用中的IP，但实际ping不通，而且connect_alarm_num（非法使用天数）有残留数据的测试用例

预期结果：disconnect_alarm_num（无法连接天数）+1、connect_alarm_num（非法使用天数）=0

还可以增加测试用例

① 3.3.3.3是被定义为空闲的IP，且实际ping不通，但connect_alarm_num（非法使用天数）有非0数据的测试用例

预期结果：connect_alarm_num（非法使用天数）=0、disconnect_alarm_num（无法连接天数）=0

②4.4.4.4是被定义为使用中的IP，且实际可以ping通，但disconnect_alarm_num（无法连接天数）有残留数据的测试用例

预期结果：disconnect_alarm_num（无法连接天数）=0、connect_alarm_num（非法使用天数）=0

from django.test import TestCase, TransactionTestCase
from mysiteapp import views
from mysiteapp.models import ipaddr_info

# Create your tests here.
class start_ping_test(TransactionTestCase):
    def setUp(self):
        ipaddr_info.objects.create(ipaddr="1.1.1.1", ipstatus="0", disconnect_alarm_num=5)
        ipaddr_info.objects.create(ipaddr="2.2.2.2", ipstatus="1", connect_alarm_num=5)
        startip='1.1.1.1'
        endip='2.2.2.2'
        stip = startip.split('.')
        enip = endip.split('.')
        tmp_ip = stip
        for i in range(int(stip[3]), int(enip[3]) + 1):
            tmp_ip[3] = str(i)
            ip = '.'.join(tmp_ip)
            ipobj = ipaddr_info.objects.get(ipaddr=ip)
            print("这是开始：", ipobj.ipaddr, ipobj.ipstatus, ipobj.disconnect_alarm_num, ipobj.connect_alarm_num)

    def test_ping(self):
        startip = '1.1.1.1'
        endip = '2.2.2.2'
        views.start_ping(startip, endip)  #TestCase情况下，从主函数进去，调用子函数的时候查无数据
        # views.ip_status_alarm(startip)      #直接从子函数进去，才有数据
        # views.ip_status_alarm(endip)

    def tearDown(self):
        for item in ipaddr_info.objects.all():
            print("这是结束：", item.ipaddr, item.ipstatus, item.disconnect_alarm_num, item.connect_alarm_num)

执行结果：