cros_test_ready: initial version
Add a new tool to check if a DUT is ready to be used for testing.
BUG=b:284903141
TEST=sudo emerge-hana cros-test-ready
Cq-Depend: chromium:4702115
Cq-Depend: chromium:4700387
Change-Id: I4cda386dcc2afcafd40d4ebe47cd15aacb5c9b76
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/dev-util/+/4702272
Tested-by: Seewai Fu <seewaifu@google.com>
Reviewed-by: Derek Beckett <dbeckett@chromium.org>
Commit-Queue: Seewai Fu <seewaifu@google.com>
diff --git a/src/chromiumos/test/check/cmd/cros_test_ready/main.go b/src/chromiumos/test/check/cmd/cros_test_ready/main.go
new file mode 100644
index 0000000..d0c4779
--- /dev/null
+++ b/src/chromiumos/test/check/cmd/cros_test_ready/main.go
@@ -0,0 +1,211 @@
+// Copyright 2023 The ChromiumOS Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+// Package main implements the cros-test-finder for finding tests based on tags.
+package main
+
+import (
+ "context"
+ "crypto/md5"
+ "flag"
+ "fmt"
+ "io"
+ "log"
+ "os"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/golang/protobuf/jsonpb"
+ "github.com/google/subcommands"
+
+ "go.chromium.org/chromiumos/config/go/test/api"
+)
+
+const (
+ defaultConfigPath = "/usr/local/etc/cros_test_ready_config.jsonpb"
+)
+
+// createLogFile creates a file and its parent directory for logging purpose.
+func createLogFile(fullPath string) (*os.File, error) {
+ if fullPath == "" {
+ t := time.Now()
+ fullPath = fmt.Sprintf("/var/tmp/cros_test_ready/log_%s.txt", t.Format("20060102150405"))
+ }
+ if err := os.MkdirAll(fullPath, 0755); err != nil {
+ return nil, fmt.Errorf("failed to create directory %v: %w", fullPath, err)
+ }
+
+ logFullPathName := filepath.Join(fullPath, "log.txt")
+
+ // Log the full output of the command to disk.
+ logFile, err := os.Create(logFullPathName)
+ if err != nil {
+ return nil, fmt.Errorf("failed to create file %v: %w", fullPath, err)
+ }
+ return logFile, nil
+}
+
+// newLogger creates a logger. Using go default logger for now.
+func newLogger(logFile *os.File) *log.Logger {
+ var writer io.Writer = os.Stderr
+ if logFile != nil {
+ writer = io.MultiWriter(logFile, os.Stderr)
+ }
+ return log.New(writer, "", log.LstdFlags|log.LUTC)
+}
+
+// readInput reads a CrosTestReadyConfig jsonproto file and returns a pointer to RunTestsRequest.
+func readInput(fileName string) (*api.CrosTestReadyConfig, error) {
+ f, err := os.Open(fileName)
+ if err != nil {
+ return nil, fmt.Errorf("fail to read file %v: %v", fileName, err)
+ }
+ req := api.CrosTestReadyConfig{}
+ umrsh := jsonpb.Unmarshaler{}
+ umrsh.AllowUnknownFields = true
+ if err := umrsh.Unmarshal(f, &req); err != nil {
+ return nil, fmt.Errorf("fail to unmarshal file %v: %v", fileName, err)
+ }
+ return &req, nil
+}
+
+// Version is the version info of this command. It is filled in during emerge.
+var Version = "<unknown>"
+
+type src2DestDir struct {
+ src string
+ destDir string
+}
+
+type fileMappings struct {
+ mapping []src2DestDir
+}
+
+func (fm *fileMappings) String() string {
+ return fmt.Sprintf("%v", fm.mapping)
+}
+
+func (fm *fileMappings) Set(value string) error {
+ mapping := strings.Split(value, ":")
+ if len(mapping) != 2 {
+ return fmt.Errorf("invalid string for file mapping: %q", mapping)
+ }
+ fm.mapping = append(fm.mapping, src2DestDir{src: mapping[0], destDir: mapping[1]})
+ return nil
+}
+
+// checkCmd implements subcommands.Command to check
+// if a DUT is test ready to be used for testing.
+type checkCmd struct {
+ config string
+ logFile string
+ exitOnError bool
+ logger *log.Logger
+}
+
+// newCheckCmd returns a new checkCmd that will check
+// if a DUT is test ready to be used for testing.
+func newCheckCmd() *checkCmd {
+ return &checkCmd{}
+}
+
+func (*checkCmd) Name() string { return "check" }
+func (*checkCmd) Synopsis() string { return "check" }
+func (*checkCmd) Usage() string {
+ return `Usage: generate [flag]...
+
+Description:
+ Check whether a DUT is test ready.
+ It will exit with status 0 if all checks are passed.
+ Otherwise, it will exit with status 1.
+
+Flag:
+`
+}
+
+func (c *checkCmd) SetFlags(f *flag.FlagSet) {
+ f.StringVar(&c.config,
+ "config", defaultConfigPath,
+ "Specify the jsonpb configuration file for what are needed to be checked on the DUT.")
+ f.StringVar(&c.logFile,
+ "log", "",
+ "Specify the log file path (default: /var/tmp/cros_test_ready/log_<YYYYmmDDHHMMSS>.txt)")
+ f.BoolVar(&c.exitOnError,
+ "exit_on_error",
+ true,
+ "Specify whether cros_test_ready should exit when it encounters the first error.")
+
+}
+
+func (c *checkCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
+ fullLog, err := createLogFile(c.logFile)
+ if err != nil {
+ c.logger = newLogger(nil)
+ c.logger.Printf("Failed to create log file %s; use stderr only: %v", c.logFile, err)
+ } else {
+ c.logger = newLogger(fullLog)
+ defer fullLog.Close()
+ }
+
+ checkConfig, err := readInput(c.config)
+ if err != nil {
+ c.logger.Fatalf("Failed to read configuration %s: %v", c.config, err)
+ }
+ hasError := false
+
+ for _, fileChecksum := range checkConfig.GetChecksums() {
+ checksum, err := md5Str(fileChecksum.GetKey())
+ if err != nil {
+ c.logger.Printf("Failed to get checksum for file %s: %v", fileChecksum.GetKey(), err)
+ hasError = true
+ }
+ if checksum != fileChecksum.GetValue() {
+ c.logger.Printf("Failed to get expected checksum for file %s: got %q; wanted %q",
+ fileChecksum.GetKey(), checksum, fileChecksum.GetValue())
+ hasError = true
+ }
+ if c.exitOnError && hasError {
+ return subcommands.ExitFailure
+ }
+ }
+ return subcommands.ExitSuccess
+}
+
+func md5Str(filename string) (string, error) {
+ f, err := os.Open(filename)
+ if err != nil {
+ return "", fmt.Errorf("failed to open file %s: %v", filename, err)
+ }
+ defer f.Close()
+
+ h := md5.New()
+ if _, err := io.Copy(h, f); err != nil {
+ return "", fmt.Errorf("failed to calculate checksum for file %s: %v", filename, err)
+ }
+
+ return fmt.Sprintf("%x", h.Sum(nil)), nil
+}
+
+func mainInternal(ctx context.Context) int {
+ subcommands.Register(subcommands.HelpCommand(), "")
+ subcommands.Register(subcommands.FlagsCommand(), "")
+ subcommands.Register(subcommands.CommandsCommand(), "")
+ subcommands.Register(newCheckCmd(), "")
+
+ version := flag.Bool("version", false, "print version and exit")
+
+ flag.Parse()
+
+ if *version {
+ fmt.Printf("cros_test_ready version %s\n", Version)
+ return 0
+ }
+
+ return int(subcommands.Execute(ctx))
+}
+
+func main() {
+ os.Exit(mainInternal(context.Background()))
+}
diff --git a/src/chromiumos/test/check/python/cros_test_ready_config_generator.py b/src/chromiumos/test/check/python/cros_test_ready_config_generator.py
new file mode 100755
index 0000000..011bc53
--- /dev/null
+++ b/src/chromiumos/test/check/python/cros_test_ready_config_generator.py
@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+# Copyright 2023 The ChromiumOS Authors
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Generate create_test_ready.jsonpb for checking DUT readiness for testing."""
+
+import argparse
+import hashlib
+import os
+import sys
+
+from chromiumos.test.api import cros_test_ready_cli_pb2 as test_ready_pb
+from google.protobuf.json_format import MessageToJson
+import six
+
+
+# If running in Autotest dir, keep this.
+os.environ["PY_VERSION"] = "3"
+
+# NOTE: this MUST be run in Python3, if we get configured back to PY2, exit.
+if six.PY2:
+ sys.exit(1)
+
+
+def autotest_checksum_dirs():
+ """Return a mapping between source and destination paths of directories.
+
+ This function return a mapping between source and destination paths
+ of autotest directories.
+ """
+
+ return { "/usr/local/autotest": "/usr/local/autotest" }
+
+
+def default_checksum_files():
+ """Return a mapping between source and destination paths of files.
+
+ For test image only sources that are not in /user/local will be
+ moved to /user/local in a DUT.
+ """
+
+ return {
+ "/usr/local/autotest/bin/autotest":
+ "/usr/local/autotest/bin/autotest",
+
+ "/usr/libexec/tast/bundles/local/cros":
+ "/usr/local/libexec/tast/bundles/local/cros",
+
+ "/usr/bin/local_test_runner":
+ "/usr/local/bin/local_test_runner",
+ }
+
+
+def parse_local_arguments(args):
+ """Parse the CLI."""
+ parser = argparse.ArgumentParser(
+ description="Create config file for checking if a DUT is test ready."
+ )
+ parser.add_argument(
+ "-src_root",
+ dest="src_root",
+ default=None,
+ help="Path to root directory of the source files or directories.",
+ )
+ parser.add_argument(
+ "-output_file",
+ dest="output_file",
+ default=None,
+ help="Where to write the cros_test_ready.jsonpb.",
+ )
+ return parser.parse_args(args)
+
+
+def md5(filename):
+ with open(filename, "rb") as f:
+ content = f.read()
+ readable_hash = hashlib.md5(content).hexdigest()
+ return readable_hash
+
+
+def serialize_checksum_from_file(src, dest):
+ """Calculate checksum of a file."""
+ try:
+ md5_value = md5(src)
+ return test_ready_pb.CrosTestReadyConfig.KeyValue(
+ key=dest, value=md5_value
+ )
+ except Exception as e:
+ raise Exception(
+ "Failed to calculate checksum for file %s: %s" % (src, e)
+ )
+
+
+def serialize_autotest_checksum_from_dir(src_dir, dest_dir):
+ """Calculate checksums of non-hidden _python_ files under a directory."""
+ results = []
+ try:
+ for root, _, filenames in os.walk(src_dir):
+ for filename in filenames:
+ # Ignore hidden files.
+ if filename.startswith("."):
+ continue
+ # Only check for python files for autotest.
+ if not filename.endswith(".py"):
+ continue
+ src_file_path = os.path.join(root, filename)
+ dest_file_path = src_file_path.replace(src_dir, dest_dir)
+ results.append(
+ serialize_checksum_from_file(
+ src_file_path, dest_file_path
+ )
+ )
+ except Exception as e:
+ raise Exception(
+ "Failed to calculate checksum for files under %s: %s"
+ % (src_dir, e)
+ )
+ return results
+
+
+def serialize_checksum(src_root):
+ """Return a serialized TestCaseInfo obj."""
+ checksums = []
+ filenames = default_checksum_files()
+ for src, dest in filenames.items():
+ checksums.append(serialize_checksum_from_file(
+ src_root + src, dest))
+ # Calculate checksum for python files under autotest directories.
+ dirs = autotest_checksum_dirs()
+ for src_dir, dest_dir in dirs.items():
+ checksums_for_dir = serialize_autotest_checksum_from_dir(
+ src_root + src_dir, dest_dir)
+ checksums = checksums + checksums_for_dir
+ return test_ready_pb.CrosTestReadyConfig(checksums=checksums)
+
+
+def main():
+ """Generate the metadata, and if an output path is given, save it."""
+ args = parse_local_arguments(sys.argv[1:])
+ serialized = serialize_checksum(args.src_root)
+ json_obj = MessageToJson(serialized)
+
+ if args.output_file:
+ with open(args.output_file, "w", encoding="utf-8") as wf:
+ wf.write(json_obj)
+
+
+if __name__ == "__main__":
+ main()
