python考核试题

1、请编写装饰器runtime计算方法执行时间

实现方法：

def runtime(func):
    from time import time
    def wrapper(*args, **kwargs):
        # 定义开始和结束时间,把func放中间取得返回值
        start = time()
        func_return = func(*args, **kwargs)
        end = time()
        # 打印方法名和执行时间
        print(f'{func.__name__}() execute time: {end - start}s')
        # 返回func的返回值
        return func_return
    # 返回嵌套的函数
    return wrapper
@runtime
def say_hello():
    print("hello!")
print(say_hello())

import datetime
import functools
def runtime(func):
    @functools.wraps(func)
    def wrapper():
        starttime = datetime.datetime.now()
        print(starttime)
        return func()
    return wrapper
@runtime
def say_hello():
    print("hello!")
say_hello()

2、补全代码

请为 calculate 添加一个新方法run，实现通过传入方法名和参数即可调用对应方法，如下

class calculate:
   def run(self, value, *args):
       print(eval('self.{}(*args)'.format(value)))
   def sum(self, a, b):
       return a+b
   def minus(self, a, b):
       return a-b
   def sum2(self, a, b, c):
       return a+b+c
   def multi(self, a, b):
      return a*b
calc = calculate()
calc .run("sum", 1, 2)
calc .run("sum2", 1, 2, 3)

class Calculate:
    def sum(self, a, b):
        return a + b
    def minus(self, a, b):
        return a - b
    def sum2(self, a, b, c):
        return a + b + c
    def multi(self, a, b):
        return a * b
    def run(self, method, *args):
        if method in dir(Calculate):
            try:
                re = eval('self.{}(*args)'.format(method))
            except TypeError:
                print('参数错误')
                return
            return re
        else:
            print('没有该函数')
            return None
if __name__ == '__main__':
    calc = Calculate()
    print(calc.run("sum", 1, 2))
    print(calc.run("sum2", 1, 2, 3))

3、 列表推导式
members = [(‘liming’, ‘man’), (‘zhaoli’, ‘woman’), (‘zhang3’, ‘unknow’), (‘humei’, ‘woman’)]
编写代码列出所有的woman

members = [('liming', 'man'), ('zhaoli', 'woman'), ('zhang3', 'unknow'), ('humei', 'woman')]
a = []
[a.append(members[i][0]) for i in range(4) if  members[i][1] == 'woman']
print(a)

4、 使用python编写socket程序，实现两个机器间传输文件

server_socket.py
#!/usr/bin/python3
import socket
s = socket.socket() #默认走TCP协议、IPV4地址格式
ip_port = ('172.27.144.28',9999)
s.bind(ip_port)
s.listen(2)
conn, address = s.accept()
print("开始接收文件")
while True:
    with open('/root/sock/file','ab') as f:
        data = conn.recv(1024)
        if data == b'quit':
            break
        f.write(data)
        conn.send('success'.encode())
print("文件接收完成")
s.close()
client_socket.py
#!/usr/bin/python3
import socket
s = socket.socket()
ip_port = ('172.27.144.28', 9999)
s.connect(ip_port)
with open('/root/sock/test.py') as f:
    for i in f:
        i = i.encode()
        s.send(i)
        data = s.recv(1024)
        if data != b'success':
            break
s.send('quit'.encode())

5、写一个能产生 100 个僵尸进程的程序
https://www.cnblogs.com/lemon-le/p/8125201.html
https://blog.csdn.net/Freshduke/article/details/111544319

#/usr/bin/python3
import os
import logging
import time
from multiprocessing import Process, current_process
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)- 15s - %(levelname)s -%(message)s'
)
def run():
    logging.info('exit son process %s', current_process().pid)
    os._exit(3)
for i in range (1,100):
    p = Process(target=run)
    p.start()
    print("start subprocess %d", os.getpid())
    time.sleep(1)

6、 使用python编写程序，获取机器上物理网卡的信息，包括ip、mac、网卡名，以json格式输出
下面是一段网卡信息的输出

import netifaces
import json
for name in netifaces.interfaces():
    info = netifaces.ifaddresses(name)
    for key in info.keys():
        if key == 2:
            info['IPV4'] = info.pop(key)
        elif key == 10:
            info['IPV6'] = info.pop(key)
        elif key == 17:
            info['MAC'] = info.pop(key)
    new_dict = {}
    new_dict[name] = info
    new_info = json.dumps(new_dict, sort_keys=True, indent=4, separators=(',', ':'))
    print('--wait--')
    print(new_info)

7、首先编写一个tcp套接字服务端程序，能监听在指定端口（如 889，可通过命令行参数指定），然后编写一个客户端程序连接该服务端程序（要求客户端能逐渐消耗掉其所在机器的端口，每 0.01 秒增加一个线程去连接服务端）
服务器

import socket
import threading
import time
def thread_job():
    while True:
        print(threading.current_thread())
        print(threading.active_count())
        time.sleep(1000)
def main():
    server = socket.socket()
    server.bind(('172.27.144.19', 10100))
    server.listen(10)
    print('wait for connection.')
    while True:
        conn,addr = server.accept()
        print(conn,addr)
        print('continue.')
        thread = threading.Thread(target=thread_job, )
        thread.start()
    server.close()
if __name__ == '__main__':
    main()

客户端

def thread_job():
    print(threading.current_thread())
def main():
    while 1:
        client = socket.socket()
        client.connect(('172.27.144.19', 10100))
        thread = threading.Thread(target=thread_job,)
        thread.start()
        time.sleep(2)  // 0.1   
if __name__ == '__main__':
    main()

解法2

# !/usr/bin/python3
from socket import *
from threading import Thread
import sys
class Server:
    def __init__(self, address):
        self.address = address
        self.__create_socket()
        self.__bind()
    def __create_socket(self):
        self.sockfd = socket()
        self.sockfd.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    def __bind(self):
        self.sockfd.bind(self.address)
        self.ip = self.address[0]
        self.port = self.address[1]
    def serve_start(self):
        self.sockfd.listen(50)
        print('Now listen port %d' % self.port)
        while True:
            connfd, addr = self.sockfd.accept()
            print('connect ', addr)
            client = Thread(target=self.__handle, args=(connfd,), daemon=True)
            client.start()
    def __handle(self, connfd):
        pass
if __name__ == '__main__':
    args = sys.argv[1:]
    host = args[args.index('--host') + 1] if '--host' in args else '172.27.144.19'
    port = int(args[args.index('--port') + 1]) if '--port' in args else 7408
    server = Server((host, port))
    server.serve_start()

# !/usr/bin/python3
from socket import *
from threading import Thread
import time
ADDR = ('172.27.144.19', 7408)
def consume(sockfd):
    sockfd.connect(ADDR)
while True:
    try:
        sockfd = socket()
        t = Thread(target=consume, args=(sockfd,), daemon=True)
        t.start()
        time.sleep(0.1)
    except KeyboardInterrupt:
        print('Over.')
        break

8、编写两个程序，第一个能从 /dev/random 中读取数据，第二个能接收第一个读取的数据然后写入到 /dev/null 中，要求这两段程序运行在不同的机器上，可通过命令行参数来决定是通过 tcp 还是 udp 来传输数据，并且能分别控制数据的读取和传输速度

控制速度
from socket import *
import sys
import time
def read(sock_type='tcp', read_spead=1):
    with open('/dev/random', 'rb') as f:
        data = ''.encode()
        for i in range(read_spead):
            data += f.read(1024)
            time.sleep(1/4)
    return data
print(read())

client.py

# !/usr/bin/python3
from socket import *
import sys
class Client:
    def __init__(self, host, port):
        self.host = host
        self.port = port
    def run(self, read_spead, transport_speed):
        raise NotImplementedError
    def recv_and_write(self, conn, sock_type='tcp', write_spead=1024, transport_speed=1024):
        while True:
            with open('/dev/null', 'wb') as f:
                data = conn.recvfrom(transport_speed)[0] if sock_type == 'udp' else conn.recv(transport_speed)
                if data:
                    for i in range(0, len(data), write_spead):
                        f.write(data[i:i+write_spead])
                        print(data[i:i+write_spead])
                else:
                    return
    def close(self):
        raise NotImplementedError
class TCPClient(Client):
    def __init__(self, host, port):
        super().__init__(host, port)
        self.conn = socket()
    def run(self, write_speed, transport_speed):
        self.conn.connect((self.host, self.port))
        self.recv_and_write(self.conn, 'tcp', write_spead=write_speed, transport_speed=transport_speed)
        self.close()
    def close(self):
        self.conn.close()
class UDPClient(Client):
    def __init__(self, host, port):
        super().__init__(host, port)
        self.conn = socket(type=SOCK_DGRAM)
    def run(self, write_speed, transport_speed):
        self.conn.sendto(b'nothing', (self.host, self.port))
        self.recv_and_write(self.conn, 'udp', write_spead=write_speed, transport_speed=transport_speed)
        self.close()
    def close(self):
        self.conn.close()
if __name__ == '__main__':
    sock_type = 'udp' if len(sys.argv) > 1 and sys.argv[1].lower() == 'udp' else 'tcp'
    HOST = '172.27.144.19'
    PORT = 7074
    if sock_type == 'tcp':
        server = TCPClient(HOST, PORT)
    else:
        server = UDPClient(HOST, PORT)
    try:
        server.run(write_speed=10, transport_speed=1024)
    except KeyboardInterrupt:
        print('Over')

server.py

# !/usr/bin/python3
from socket import *
import sys
class Server:
    def __init__(self, host, port):
        self.host = host
        self.port = port
    def run(self, read_spead, transport_speed):
        raise NotImplementedError
    def read_and_send(self, conn, sock_type='tcp', addr=None, read_spead=1024, transport_speed=1024):
        try:
            while True:
                with open('/dev/random', 'rb') as f:
                    data = f.read(read_spead)
                    for i in range(0, len(data), transport_speed):
                        conn.sendto(data[i:transport_speed + i], addr) if sock_type == 'udp' else conn.send(
                            data[i:transport_speed + i])
        except BrokenPipeError:
            print('The client disconnects and the program exits.')
            return
        except ConnectionResetError:
            print('The client disconnects and the program exits.')
            return
    def close(self):
        raise NotImplementedError
class TCPServer(Server):
    def __init__(self, host, port):
        super().__init__(host, port)
        self.conn = socket()
        self.conn.bind((host, port))
    def run(self, read_spead, transport_speed):
        self.conn.listen(5)
        self.connfd, self.addr = self.conn.accept()
        self.read_and_send(self.connfd, 'tcp')
        self.close()
    def close(self):
        self.connfd.close()
        self.conn.close()
class UDPServer(Server):
    def __init__(self, host, port):
        super().__init__(host, port)
        self.conn = socket(type=SOCK_DGRAM)
        self.conn.bind((host, port))
    def run(self, read_spead, transport_speed):
        recv_data, addr = self.conn.recvfrom(1024)
        self.read_and_send(self.conn, 'udp', addr=addr, read_spead=read_spead, transport_speed=transport_speed)
        self.close()
    def close(self):
        self.conn.close()
if __name__ == '__main__':
    sock_type = 'udp' if len(sys.argv) > 1 and sys.argv[1].lower() == 'udp' else 'tcp'
    HOST = '172.27.144.19'
    PORT = 7074
    if sock_type == 'tcp':
        server = TCPServer(HOST, PORT)
    else:
        server = UDPServer(HOST, PORT)
    try:
        server.run(read_spead=1024, transport_speed=1024)
    except KeyboardInterrupt:
        print('Over')