上传新文件

tool/check_3.py

+#!/usr/bin/python3
+
+import datetime
+import time
+import subprocess
+import shlex
+
+
+logfile = "/var/log/check_3.log"
+
+
+class check_all():
+    def __init__(self, disks:list, disk_num: int, ints: list, memsize: int):
+        self.disks = disks
+        self.disk_num = disk_num
+        self.ints = ints
+        self.memsize = memsize
+
+    def sata_speed(self):
+        """
+        测试硬盘SATA接口速率
+        """
+        for i in self.disks:
+            cmd = f"smartctl -a /dev/{i} | grep -c 'SATA 3.* 6.0 Gb/s'"
+            ret = runcmd(cmd, shell=True)
+            if ret[1] or not int(ret[0].replace("\n", "")):
+                logs(f"Error:硬盘{i} 速率检测失败,sata接口没有达到6Gb,将设备进行分类.")
+                return 0
+            logs(f"info:硬盘{i} 速率检测正常,sata接口速率为6Gb!")
+        return 1
+
+    def intf(self):
+        """
+        测试网络接口
+        """
+        for i in self.ints:
+            cmd = f"ethtool {i} | egrep -c 'Speed: 1000Mb/s|Link detected: yes'"
+            ret = runcmd(cmd, shell=True)
+            if ret[1] or int(ret[0].replace("\n", "")) != 2:
+                logs(f"Error:网卡{i}没有连接网线或速率没有达到1Gb,将设备进行分类.")
+                return 0
+            logs(f"info:网卡{i}接口正常.")
+        return 1
+
+    def mem(self):
+        """
+        测试内存大小
+        """
+        cmd = "free -g | grep Mem | awk '{print $2}'"
+        ret = runcmd(cmd, shell=True)
+        if not ret[1] and int(ret[0].replace("\n", "")) >= self.memsize:
+            logs("info:内存检测正常.")
+            return 1
+        else:
+            logs("Error:内存检测失败,将设备进行分类.")
+            return 0
+
+    def hd_num(self):
+        """
+        测试HDD硬盘数量
+        """
+        n = len(disks())
+        if n == self.disk_num:
+            logs(f"info:硬盘数量为{n},符合标配.")
+            return 1
+        else:
+            logs(f"Error:硬盘数量为{n},与标配{self.disk_num}不符.")
+            return 0
+
+    def sync_time(self):
+        """
+        同步时间
+        """
+        cmd = "ntpdate ntp3.aliyun.com"
+        ret = runcmd(cmd)
+        if not ret[1]:
+            time.sleep(3)
+            runcmd("hwclock -w")
+            print("当前时间:", runcmd("hwclock -r"))
+            return 1
+        else:
+            logs("Error:时间同步失败,请检测网络.")
+            return 0
+
+    def usb(self):
+        """
+        测试usb接口
+        """
+        cmd = "udevadm info -q path -n /dev/sd[a-i]|grep usb|grep -ocE '[^/]+$'"
+        n = 0
+        while n < 3:
+            ret = runcmd(cmd, shell=True)
+            if ret[1] or not int(ret[0].replace("\n", "")):
+                print("Error: 没检测到U盘请认插入usb设备")
+                print("再次检测中...")
+                time.sleep(10)
+                n += 1
+            else:
+                logs("info:USB设备检测正常.")
+                return 1
+        logs("Error:USB设备检测失败,如果已接入usb设备,请将设备归类.")
+        return 0
+
+
+def model():
+    """
+    查看型号并返回标配内存硬盘数量
+    """
+    modes={"A000": [7, 4, 8, "t"],
+           "A001": [7, 4, 16, "t"],
+           "A002": [7, 4, 32, "t"],
+           "A003": [7, 4, 64, "t"],
+           "A004": [15, 4, 8, "t"],
+           "A005": [15, 4, 16, "t"],
+           "A006": [15, 4, 32, "t"],
+           "A007": [15, 4, 64, "t"],
+           "A100": [7, 4, 16, "w"],
+           "A101": [7, 4, 32, "w"],
+           "A102": [7, 4, 64, "w"],
+           "A103": [15, 4, 16, "w"],
+           "A104": [15, 4, 32, "w"],
+           "A105": [15, 4, 64, "w"],
+           "A200": [15, 4, 32, "w"],
+           "A201": [15, 4, 64, "w"],
+           "B000": [15, 4, 32, "w"],
+           "B001": [15, 4, 64, "w"],
+           "B100": [15, 4, 32, "w"],
+           "B101": [15, 4, 64, "w"],
+           "B200": [15, 4, 64, "w"],
+           "B300": [15, 8, 15.36, "w"],
+           "B301": [15, 8, 30.72, "w"],
+           "B400": [15, 8, 30.72, "w"],
+           "B500": [15, 8, 30.72, "w"]}
+
+    cmd = "/opt/tool/saturn-device-id -r 14 | cut -c 11-"
+    ret = runcmd(cmd, shell=True)
+    if ret[1]:
+        return 0
+    else:
+        sn = ret[0].replace("\n", "")
+        return modes[sn], sn
+
+
+def disks():
+    """
+    HDD硬盘列表
+    """
+    cmd = "lsblk -d -o NAME,RM,ROTA | grep -v NAME"
+    check_ds = []
+    ret = runcmd(cmd, shell=True)
+    if ret[1]:
+        return 0
+    else:
+        for l in str(ret[0]).split("\n"):
+            l1 = l.split()
+            if l1 and int(l1[1]) == 0 and int(l1[2]) == 1:
+                check_ds.append(l1[0])
+        return check_ds
+
+
+def logs(log: str, op="a"):
+    with open(logfile, op) as f:
+        t = str(datetime.datetime.now()).split(".")[0]
+        line = t + " " + log + "\n"
+        print(line, end="")
+        f.write(line)
+
+
+def runcmd(cmd, shell=False, encoding='utf-8'):
+    args = cmd if shell else shlex.split(cmd)
+    r = subprocess.run(args, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding=encoding)
+    return r.stdout, r.returncode
+
+
+def check_in():
+    sn = str(input("请扫码获取序列号：")).replace(" ", "")
+    if len(sn) == 31 and "-" in sn:
+        s = sn.split("-")
+        if len(s[0]) == 14 and len(s[1]) == 16:
+            return sn
+    elif len(sn) == 14:
+            return sn
+
+
+if __name__ == '__main__':
+    print("***本程序为出厂测试，继续运行会清除所有用户数据，若是误打开请按Ctrl + C退出！***")
+    sn = check_in()
+    if sn:
+        cmd = f"/opt/tool/saturn-device-id -w ucaskodcloudwanghuo {sn}"
+        ret = runcmd(cmd)
+    else:
+        print("序列号格式不正确！")
+        exit(1)
+    logs("开始检测...", op="w")
+    if not ret[1]:
+        logs(f"info:序列号{sn}写入成功.")
+        mode = model()
+        if mode[0]:
+            ints = ["eth0", "eth1"] if mode[0][3] == "w" else ["end0", "end1"]
+            mynas = check_all(disks(), mode[0][1], ints, mode[0][0])
+        else:
+            logs("型号未能识别，检查序列号是否正确。")
+            exit(1)
+        if not mynas.sata_speed():
+            exit(1)
+        if not mynas.intf():
+            exit(1)
+        if not mynas.mem():
+            exit(1)
+        if not mynas.hd_num():
+            exit(1)
+        if not mynas.usb():
+            exit(1)
+        #if not mynas.sync_time():
+        #    exit(1)
+        logs("检测通过！")
+        print("开始运行老化测试,测试时间约为10小时.")
+        ret2 = runcmd("nohup /opt/tool/emaging.py startema > /dev/null 2>&1 &", shell=True)
+        if not ret2[1]:
+            print("老化测试已开始运行,心跳灯开始绿灯闪烁\n当心跳灯绿色常亮时，为老化正常结束\n如果红色常亮或不亮时，则为异常，请查看日志：")
+        else:
+            print("老化测试运行失败")
+        print("老化日志查看：cat /var/log/check_3.log")
+        exit()
+
+    else:
+        logs("Error:序列号写入失败.")
+        exit(1)