Write end-to-end tests

Author: Andrew Orals

dgrep/main.go

@@ -13,13 +13,10 @@ type Args struct {
   Query string
 }
 
-const kListenPort = 1233
-const kRpcPort = "1234"
-
 var address = flag.String("a", "localhost", "The address of the log servent")
 
 func query(q string, addr string) {
-  c, err := rpc.DialHTTP("tcp", addr + ":" + kRpcPort)
+  c, err := rpc.DialHTTP("tcp", addr)
   if err != nil {
     log.Fatal("dialing:", err)
   }

logservent/main.go

@@ -5,14 +5,12 @@ import (
   "flag"
   "log"
   "os"
-  "os/exec"
-  "path/filepath"
   "net"
   "net/rpc"
   "net/http"
   "bufio"
   "sync"
-  "errors"
+  "strings"
   "andreworals.com/dlogger/leveled_logger"
   "andreworals.com/dlogger/chan_grep"
   "github.com/hsfzxjy/go-srpc"
@@ -24,8 +22,6 @@ import (
 
 var logger leveled_logger.LeveledLogger
 
-const rpcPort = "1234"
-const kListenPort = 1235
 var logDir = flag.String("d", "logs", "log directory")
 var debugLevel = flag.Int("v", 1, "verbosity level, 0=off 1=normal 2=debug")
 var machineAddrs []string
@@ -39,12 +35,12 @@ type GrepClientRpc string
 
 // RPC from queried log server to log server
 // Takes output from grep and sends it over a TCP connection to the caller
-func (*LogServerRpc) LoggerQuery(args Args, s *srpc.Session) error {
+func (*LogServerRpc) LoggerQuery(args chan_grep.Args, s *srpc.Session) error {
   return srpc.S( func() error {
     resultsCh := make(chan string)
     errCh := make(chan error)
 
-    go grepLogs(args, *logDir, resultsCh, errCh)
+    go chan_grep.GrepLogs(args, *logDir, resultsCh, errCh)
 
     for {
       select {
@@ -59,11 +55,12 @@ func (*LogServerRpc) LoggerQuery(args Args, s *srpc.Session) error {
   }, s, nil)
 }
 
-func serverReader(wg *sync.WaitGroup, args Args, addrStr string, listenPort int, ch chan<- string) {
+func serverReader(wg *sync.WaitGroup, args chan_grep.Args, addrStr string, ch chan<- string) {
   defer wg.Done()
 
-  c, err := rpc.DialHTTP("tcp", addrStr + ":" + rpcPort)
+  c, err := rpc.DialHTTP("tcp", addrStr)
   if err != nil {
+    logger.PrintfNormal("Could not dial logserver at %s: %s\n", addrStr, err)
     return
   }
   client := srpc.WrapClient(c)
@@ -83,7 +80,7 @@ func done(wg *sync.WaitGroup, doneCh chan<- Nothing, ch chan<- string) {
 
 // RPC from grep client to query log server
 // Calls each log server to retrieve the its logs and populates a channel to send back to the grep client over TCP
-func (*GrepClientRpc) GrepQuery(args Args, s *srpc.Session) error {
+func (*GrepClientRpc) GrepQuery(args chan_grep.Args, s *srpc.Session) error {
   logger.PrintfDebug("Received grep query: %s\n", args.Query)
   return srpc.S( func() error {
     // query the other servers
@@ -92,15 +89,15 @@ func (*GrepClientRpc) GrepQuery(args Args, s *srpc.Session) error {
     var wg sync.WaitGroup
 
     // Grep from other servers
-    for i, addrStr := range machineAddrs {
+    for _, addr := range machineAddrs {
       wg.Add(1)
-      go serverReader(&wg, args, addrStr, kListenPort + i, grepResultsCh)
+      go serverReader(&wg, args, addr, grepResultsCh)
     }
     // Grep from this server
     wg.Add(1)
     go func(wgroup *sync.WaitGroup) {
       defer wgroup.Done()
-      grepLogs(args, grepResultsCh, nil)
+      chan_grep.GrepLogs(args, *logDir, grepResultsCh, nil)
       logger.PrintfDebug("grepLogs done\n")
     }(&wg)
     // Call back when the grep calls return
@@ -132,15 +129,20 @@ func readMachineFile(fileName string) {
 
   lines := bufio.NewScanner(file) // use default ScanLines
   for lines.Scan() {
-    machineAddrs = append(machineAddrs, lines.Text())
+    str := lines.Text()
+    addr_tmp := strings.Split(str, ":")
+    if len(addr_tmp) != 2 {
+      fmt.Fprintf(os.Stderr, "Invalid address or port: %s\nFormat is <address>:<port>\n", str)
+    }
+    machineAddrs = append(machineAddrs, str)
   }
 
   logger.PrintfDebug("Addresses: %v\n", machineAddrs)
 }
 
 func usage() {
   fmt.Fprintf(os.Stderr,
-              "Usage: %s [-d <log directory>] [-v <integer debug level>] <machine file>\n",
+              "Usage: %s [-d <log directory>] [-v <integer debug level>] <port> <machine file>\n",
               os.Args[0])
   flag.PrintDefaults()
 }
@@ -152,9 +154,12 @@ func main() {
   logger.SetLevel(leveled_logger.LogLevel(*debugLevel))
   defer logger.AutoPrefix("main: ")()
 
-  if flag.NArg() == 1 {
-    readMachineFile(flag.Args()[0])
+  if flag.NArg() != 2 {
+    usage()
+    os.Exit(1)
   }
+  rpcPort := flag.Args()[0]
+  readMachineFile(flag.Args()[1])
 
   rpc.Register(new(GrepClientRpc))
   rpc.Register(new(LogServerRpc))

test_full.sh

@@ -0,0 +1,9 @@
+set -e
+
+go test andreworals.com/dlogger/chan_grep
+
+go build -o ./dgrep/dgrep ./dgrep/main.go
+go build -o ./logservent/logservent ./logservent/main.go
+
+python3 ./tests/test_dgrep.py ./logservent/logservent ./dgrep/dgrep 
+